ConversionSyncService.java 35 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817
  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. Map<String, HonorBidRecord> honorBids = honorHotStore != null
  167. ? honorHotStore.findBidsByQks(qkList)
  168. : Map.of();
  169. return new BidLookup(tencentBids, honorBids);
  170. }
  171. // ─── Step 2: 组装转化记录与回传任务 ───────────────────────────────────────
  172. private void buildRecordsAndTasks(List<PaymentInfo> payments,
  173. BidLookup lookup,
  174. List<ConversionRecord> conversionRecords,
  175. List<CallbackTask> callbackTasks,
  176. SyncResult result) {
  177. for (PaymentInfo payment : payments) {
  178. BidRecord tencentBid = lookup.tencentBids.get(payment.getQk());
  179. HonorBidRecord honorBid = lookup.honorBids.get(payment.getQk());
  180. if (tencentBid == null && honorBid == null) {
  181. conversionRecords.add(toConversionRecord(payment, null, ""));
  182. result.skipped.incrementAndGet();
  183. continue;
  184. }
  185. boolean hasTencentBid = tencentBid != null;
  186. boolean hasHonorBid = honorBid != null;
  187. boolean validTencent = hasTencentBid && "tencent".equals(tencentBid.getMedia());
  188. boolean validHonor = hasHonorBid && "honor".equals(honorBid.getMedia());
  189. if (hasTencentBid && !validTencent) {
  190. log.warn("[ConversionSync] 腾讯 bid media 非法 | qk={} | media={}", payment.getQk(), tencentBid.getMedia());
  191. }
  192. if (hasHonorBid && !validHonor) {
  193. log.warn("[ConversionSync] 荣耀 bid media 非法 | qk={} | media={}", payment.getQk(), honorBid.getMedia());
  194. }
  195. if (validTencent && validHonor) {
  196. log.warn("[ConversionSync] qk 同时命中腾讯和荣耀,跳过分发 | qk={} | act={} | date={}",
  197. payment.getQk(), payment.getAct(), payment.getDate());
  198. result.skipped.incrementAndGet();
  199. continue;
  200. }
  201. if (validTencent) {
  202. conversionRecords.add(toConversionRecord(payment, tencentBid, "tencent"));
  203. callbackTasks.add(buildTencentCallbackTask(payment, tencentBid));
  204. result.matched.incrementAndGet();
  205. log.info("[ConversionSync] 匹配到腾讯转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
  206. payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());
  207. continue;
  208. }
  209. if (!validHonor) {
  210. conversionRecords.add(toConversionRecord(payment, hasTencentBid ? tencentBid : null, hasTencentBid ? tencentBid.getMedia() : ""));
  211. result.skipped.incrementAndGet();
  212. continue;
  213. }
  214. conversionRecords.add(toHonorConversionRecord(payment, honorBid));
  215. result.matched.incrementAndGet();
  216. log.info("[ConversionSync] 匹配到荣耀转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
  217. payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());
  218. CallbackTask honorTask = buildHonorCallbackTask(payment, honorBid);
  219. if (honorTask != null) {
  220. callbackTasks.add(honorTask);
  221. }
  222. }
  223. }
  224. private CallbackTask buildTencentCallbackTask(PaymentInfo payment, BidRecord bid) {
  225. String actionType = null;
  226. if (tagEventResolver != null) {
  227. actionType = tagEventResolver.resolveActionType(bid.getTagId(), payment.getAct());
  228. }
  229. if (actionType == null || actionType.isBlank()) {
  230. actionType = tencentClient.conversionActionType(payment.getAct());
  231. log.warn("[ConversionSync] 广告位回传方式未配置, 回退act-map | qk={} | tagId={} | act={} | actionType={}",
  232. payment.getQk(), bid.getTagId(), payment.getAct(), actionType);
  233. }
  234. return CallbackTask.tencent(
  235. payment,
  236. bid,
  237. callbackDedupeKey(payment, bid, actionType),
  238. actionType
  239. );
  240. }
  241. private CallbackTask buildHonorCallbackTask(PaymentInfo payment, HonorBidRecord bid) {
  242. if (honorColdStore == null && honorTagEventResolver == null) {
  243. return null;
  244. }
  245. Integer conversionId = honorTagEventResolver != null
  246. ? honorTagEventResolver.resolveConversionId(bid.getTagId(), payment.getAct())
  247. : resolveHonorConversionIdFromDb(bid.getTagId(), payment.getAct());
  248. if (conversionId == null) {
  249. log.warn("[ConversionSync] 荣耀 conversionId 未配置 | qk={} | tagId={} | act={}",
  250. payment.getQk(), bid.getTagId(), payment.getAct());
  251. return null;
  252. }
  253. return CallbackTask.honor(
  254. payment,
  255. bid,
  256. honorCallbackDedupeKey(payment, bid, conversionId),
  257. conversionId
  258. );
  259. }
  260. private Integer resolveHonorConversionIdFromDb(String tagId, int baiduAct) {
  261. if (honorColdStore == null) {
  262. return null;
  263. }
  264. com.adx.tencent.honor.model.HonorTagEventRecord exact = honorColdStore.getTagEvent(tagId, baiduAct);
  265. if (exact != null) {
  266. return exact.getConversionId();
  267. }
  268. com.adx.tencent.honor.model.HonorTagEventRecord fallback = honorColdStore.getTagEvent(tagId, 0);
  269. return fallback != null ? fallback.getConversionId() : null;
  270. }
  271. // ─── Step 3: 统一落百度转化明细 ───────────────────────────────────────────
  272. private void saveConversions(List<ConversionRecord> records) {
  273. if (coldStore == null || records.isEmpty()) {
  274. return;
  275. }
  276. try {
  277. coldStore.saveConversions(records);
  278. } catch (Exception e) {
  279. log.error("[ConversionSync] batch save conversions failed: size={}, {}", records.size(), e.getMessage(), e);
  280. }
  281. }
  282. // ─── Step 4: 回传前去重 ─────────────────────────────────────────────────
  283. private List<CallbackTask> filterPendingCallbacks(List<CallbackTask> callbackTasks, SyncResult result) {
  284. Set<String> existing = new HashSet<>();
  285. existing.addAll(collectTencentSuccessfulCallbackKeys(callbackTasks));
  286. existing.addAll(collectHonorSuccessfulCallbackKeys(callbackTasks));
  287. List<CallbackTask> pending = new ArrayList<>(callbackTasks.size());
  288. for (CallbackTask task : callbackTasks) {
  289. if (existing.contains(task.callbackKey)) {
  290. result.alreadySent.incrementAndGet();
  291. } else {
  292. pending.add(task);
  293. }
  294. }
  295. return pending;
  296. }
  297. private Set<String> collectTencentSuccessfulCallbackKeys(List<CallbackTask> callbackTasks) {
  298. if (coldStore == null) {
  299. return Set.of();
  300. }
  301. List<String> callbackKeys = new ArrayList<>();
  302. List<String> legacyKeys = new ArrayList<>();
  303. List<Integer> eventTypes = new ArrayList<>();
  304. for (CallbackTask task : callbackTasks) {
  305. if (!task.isTencent()) {
  306. continue;
  307. }
  308. callbackKeys.add(task.callbackKey);
  309. legacyKeys.add(legacyCallbackDedupeKey(task.payment, "tencent"));
  310. eventTypes.add(task.payment.getAct());
  311. }
  312. if (callbackKeys.isEmpty()) {
  313. return Set.of();
  314. }
  315. try {
  316. return new HashSet<>(coldStore.successfulCallbackKeysWithLegacy(callbackKeys, legacyKeys, eventTypes));
  317. } catch (Exception e) {
  318. log.error("[ConversionSync] batch dedup check error: size={}, {}", callbackKeys.size(), e.getMessage(), e);
  319. return Set.of();
  320. }
  321. }
  322. private Set<String> collectHonorSuccessfulCallbackKeys(List<CallbackTask> callbackTasks) {
  323. if (honorColdStore == null) {
  324. return Set.of();
  325. }
  326. Set<String> keys = new HashSet<>();
  327. for (CallbackTask task : callbackTasks) {
  328. if (task.isHonor()) {
  329. keys.add(task.callbackKey);
  330. }
  331. }
  332. if (keys.isEmpty()) {
  333. return Set.of();
  334. }
  335. try {
  336. return honorColdStore.successfulCallbackKeys(keys);
  337. } catch (Exception e) {
  338. log.error("[ConversionSync] honor dedup check error: size={}, {}", keys.size(), e.getMessage(), e);
  339. return Set.of();
  340. }
  341. }
  342. // ─── Step 5: 并发回传媒体 ───────────────────────────────────────────────
  343. private List<CallbackDispatchResult> dispatchCallbacks(List<CallbackTask> tasks,
  344. ExecutorService executor,
  345. SyncResult result) {
  346. List<CompletableFuture<CallbackDispatchResult>> futures = new ArrayList<>(tasks.size());
  347. for (CallbackTask task : tasks) {
  348. futures.add(CompletableFuture.supplyAsync(() -> sendCallback(task), executor));
  349. }
  350. List<CallbackDispatchResult> results = new ArrayList<>(tasks.size());
  351. for (CompletableFuture<CallbackDispatchResult> future : futures) {
  352. CallbackDispatchResult dispatchResult = future.join();
  353. if (dispatchResult == null || dispatchResult.record == null) {
  354. continue;
  355. }
  356. if (dispatchResult.record.isOk()) {
  357. result.sent.incrementAndGet();
  358. } else {
  359. result.failed.incrementAndGet();
  360. }
  361. results.add(dispatchResult);
  362. }
  363. return results;
  364. }
  365. private CallbackDispatchResult sendCallback(CallbackTask task) {
  366. if (task.isTencent()) {
  367. if (shouldDeductTencentCallback(task)) {
  368. MediaCallbackRecord deductedRecord = buildDeductedCallbackRecord(task);
  369. return CallbackDispatchResult.tencent(task, deductedRecord);
  370. }
  371. String platform = resolvePlatform(task.tencentBid);
  372. MediaCallbackRecord callbackRecord = tencentClient.sendConversionEvent(
  373. task.tencentBid, task.payment, platform, task.tencentActionType);
  374. callbackRecord.setDedupeKey(task.callbackKey);
  375. return CallbackDispatchResult.tencent(task, callbackRecord);
  376. }
  377. if (task.isHonor()) {
  378. HonorMediaCallbackRecord honorRecord = honorClient.sendConversion(
  379. task.honorBid, task.payment, task.honorConversionId);
  380. return CallbackDispatchResult.honor(task, toCallbackRecord(task, honorRecord));
  381. }
  382. return null;
  383. }
  384. // ─── Step 6: 落媒体回传结果 ─────────────────────────────────────────────
  385. private void saveCallbackResults(List<CallbackDispatchResult> results) {
  386. if (results.isEmpty()) {
  387. return;
  388. }
  389. List<MediaCallbackRecord> tencentRecords = new ArrayList<>();
  390. List<HonorMediaCallbackRecord> honorRecords = new ArrayList<>();
  391. for (CallbackDispatchResult result : results) {
  392. if (result.isHonor()) {
  393. honorRecords.add(toHonorCallbackRecord(result.task, result.record));
  394. } else {
  395. tencentRecords.add(result.record);
  396. }
  397. }
  398. if (coldStore != null && !tencentRecords.isEmpty()) {
  399. try {
  400. coldStore.saveMediaCallbacks(tencentRecords);
  401. } catch (Exception e) {
  402. log.error("[ConversionSync] batch save callbacks failed: size={}, {}", tencentRecords.size(), e.getMessage(), e);
  403. }
  404. }
  405. if (honorColdStore != null && !honorRecords.isEmpty()) {
  406. try {
  407. honorColdStore.saveMediaCallbacks(honorRecords);
  408. } catch (Exception e) {
  409. log.error("[ConversionSync] batch save honor callbacks failed: size={}, {}", honorRecords.size(), e.getMessage(), e);
  410. }
  411. }
  412. }
  413. // ─── 工具方法 ────────────────────────────────────────────────────────────
  414. private static String resolvePlatform(BidRecord bid) {
  415. if (bid.getPlatform() != null && !bid.getPlatform().isBlank()) {
  416. return bid.getPlatform().toLowerCase();
  417. }
  418. if (bid.getMediaParams() == null) {
  419. return "android";
  420. }
  421. String platform = bid.getMediaParams().get("baidu_platform");
  422. if (platform != null && !platform.isBlank()) {
  423. return platform.toLowerCase();
  424. }
  425. String idfa = bid.getMediaParams().get("idfa");
  426. if (idfa != null && !idfa.isBlank()) {
  427. return "ios";
  428. }
  429. return "android";
  430. }
  431. private static ConversionRecord toConversionRecord(PaymentInfo payment, BidRecord bid, String media) {
  432. ConversionRecord record = new ConversionRecord();
  433. record.setDedupeKey(conversionDedupeKey(payment));
  434. record.setQk(payment.getQk());
  435. record.setMedia(media);
  436. record.setTagId(bid != null ? bid.getTagId() : null);
  437. record.setDate(payment.getDate());
  438. record.setAppSid(payment.getAppSid());
  439. record.setCustomerName(payment.getCustomerName());
  440. record.setDeviceId(payment.getDeviceId());
  441. record.setConv(payment.getConv());
  442. record.setPayment(payment.getPayment());
  443. record.setGmv(payment.getGmv());
  444. record.setAct(payment.getAct());
  445. record.setTu(payment.getTu());
  446. record.setClkTime(payment.getClkTime());
  447. record.setCreatedAt(Instant.now());
  448. return record;
  449. }
  450. private static ConversionRecord toHonorConversionRecord(PaymentInfo payment, HonorBidRecord bid) {
  451. ConversionRecord record = new ConversionRecord();
  452. record.setDedupeKey(conversionDedupeKey(payment));
  453. record.setQk(payment.getQk());
  454. record.setMedia("honor");
  455. record.setTagId(bid != null ? bid.getTagId() : null);
  456. record.setDate(payment.getDate());
  457. record.setAppSid(payment.getAppSid());
  458. record.setCustomerName(payment.getCustomerName());
  459. record.setDeviceId(payment.getDeviceId());
  460. record.setConv(payment.getConv());
  461. record.setPayment(payment.getPayment());
  462. record.setGmv(payment.getGmv());
  463. record.setAct(payment.getAct());
  464. record.setTu(payment.getTu());
  465. record.setClkTime(payment.getClkTime());
  466. record.setCreatedAt(Instant.now());
  467. return record;
  468. }
  469. static String conversionDedupeKey(PaymentInfo payment) {
  470. return String.format("%s:%s:%s:%d",
  471. payment.getQk(),
  472. payment.getDate(),
  473. payment.getDeviceId(),
  474. payment.getAct());
  475. }
  476. static String callbackDedupeKey(PaymentInfo payment, BidRecord bid, String actionType) {
  477. return String.format("%s:%s:%s:%s:%s:%d:%s",
  478. bid.getMedia(),
  479. payment.getQk(),
  480. payment.getDate(),
  481. payment.getDeviceId(),
  482. bid.getTagId(),
  483. payment.getAct(),
  484. actionType);
  485. }
  486. static String legacyCallbackDedupeKey(PaymentInfo payment, String media) {
  487. return String.format("%s:%s:%s:%s:%d:%d",
  488. media,
  489. payment.getQk(),
  490. payment.getDate(),
  491. payment.getDeviceId(),
  492. payment.getAct(),
  493. payment.getAct());
  494. }
  495. static String honorCallbackDedupeKey(PaymentInfo payment, HonorBidRecord bid, int conversionId) {
  496. return String.format("honor:%s:%s:%s:%s:%d:%d",
  497. payment.getQk(),
  498. payment.getDate(),
  499. payment.getDeviceId(),
  500. bid.getTagId(),
  501. payment.getAct(),
  502. conversionId);
  503. }
  504. private boolean shouldDeductTencentCallback(CallbackTask task) {
  505. if (!task.isTencent() || task.tencentBid == null) {
  506. return false;
  507. }
  508. int deductionRate = resolveDeductionRate(task.tencentBid);
  509. if (task.payment.getAct() == 2001 || deductionRate <= 0) {
  510. return false;
  511. }
  512. if (!isNewTrackingLink(task.tencentBid)) {
  513. return false;
  514. }
  515. if (!hotStore.markConversionSeen(task.callbackKey)) {
  516. log.info("[ConversionSync] 腾讯转化已见过,跳过扣量计数 | callbackKey={} | qk={} | accountId={}",
  517. task.callbackKey, task.payment.getQk(), safe(task.tencentBid.getAccountId()));
  518. return false;
  519. }
  520. RedisHotStore.DeductionDecision decision = hotStore.evaluateDeduction(
  521. task.tencentBid.getAccountId(),
  522. task.payment.getAct(),
  523. resolveTrackingVersion(task.tencentBid),
  524. task.payment.getDate(),
  525. deductionRate
  526. );
  527. boolean deduct = decision.isShouldDeduct();
  528. log.info("[ConversionSync] 腾讯转化扣量判定 | qk={} | act={} | accountId={} | version={} | rate={} | totalSeen={} | totalDeducted={} | callbackKey={} | deduct={}",
  529. task.payment.getQk(), task.payment.getAct(), safe(task.tencentBid.getAccountId()),
  530. resolveTrackingVersion(task.tencentBid), deductionRate,
  531. decision.getTotalSeen(), decision.getTotalDeducted(), task.callbackKey, deduct);
  532. return deduct;
  533. }
  534. private static boolean isNewTrackingLink(BidRecord bid) {
  535. if (bid == null || bid.getMediaParams() == null) {
  536. return false;
  537. }
  538. String flag = bid.getMediaParams().get(BidRecord.NEW_LINK_FLAG_KEY);
  539. if (flag != null && Boolean.parseBoolean(flag)) {
  540. return true;
  541. }
  542. String version = bid.getMediaParams().get(BidRecord.TRACKING_VERSION_KEY);
  543. return "v2".equalsIgnoreCase(version)
  544. || "v3".equalsIgnoreCase(version)
  545. || "v4".equalsIgnoreCase(version);
  546. }
  547. private MediaCallbackRecord buildDeductedCallbackRecord(CallbackTask task) {
  548. MediaCallbackRecord record = new MediaCallbackRecord();
  549. Instant now = Instant.now();
  550. record.setDedupeKey(task.callbackKey);
  551. record.setMedia("tencent");
  552. record.setQk(task.payment.getQk());
  553. String callbackUrl = task.tencentBid.getMediaParams() != null
  554. ? task.tencentBid.getMediaParams().get("callback")
  555. : null;
  556. record.setCallbackUrl(callbackUrl != null ? callbackUrl : "");
  557. record.setEventType(task.payment.getAct());
  558. record.setEventTimeMs(now.toEpochMilli());
  559. record.setPurchase(task.payment.getPayment());
  560. record.setStatus(0);
  561. record.setOk(false);
  562. record.setAttempt(1);
  563. record.setErrorMessage("deducted_by_new_link_rate");
  564. record.setDispatchStatus(MediaCallbackRecord.DISPATCH_STATUS_DEDUCTED);
  565. record.setTrackingVersion(resolveTrackingVersion(task.tencentBid));
  566. record.setCreatedAt(now);
  567. int deductionRate = resolveDeductionRate(task.tencentBid);
  568. log.info("[ConversionSync] 腾讯转化扣量 | qk={} | act={} | rate={} | tagId={} | accountId={} | callbackKey={}",
  569. task.payment.getQk(), task.payment.getAct(), deductionRate,
  570. task.tencentBid.getTagId(), safe(task.tencentBid.getAccountId()), task.callbackKey);
  571. return record;
  572. }
  573. private static String resolveTrackingVersion(BidRecord bid) {
  574. if (bid == null || bid.getMediaParams() == null) {
  575. return "v1";
  576. }
  577. String version = bid.getMediaParams().get(BidRecord.TRACKING_VERSION_KEY);
  578. return (version == null || version.isBlank()) ? "v1" : version;
  579. }
  580. private int resolveDeductionRate(BidRecord bid) {
  581. if (bid == null || bid.getMediaParams() == null) {
  582. return 0;
  583. }
  584. String rawRate = bid.getMediaParams().get(BidRecord.DEDUCTION_RATE_KEY);
  585. if (rawRate != null && !rawRate.isBlank()) {
  586. try {
  587. return sanitizeDeductionRate(Integer.parseInt(rawRate.trim()));
  588. } catch (NumberFormatException ignored) {
  589. }
  590. }
  591. return 0;
  592. }
  593. private static String safe(String value) {
  594. return value == null ? "" : value;
  595. }
  596. private static int sanitizeDeductionRate(int rate) {
  597. if (rate < 0) {
  598. return 0;
  599. }
  600. return Math.min(rate, 100);
  601. }
  602. private static HonorMediaCallbackRecord toHonorCallbackRecord(CallbackTask task, MediaCallbackRecord record) {
  603. HonorMediaCallbackRecord callback = new HonorMediaCallbackRecord();
  604. callback.setDedupeKey(record.getDedupeKey());
  605. callback.setQk(record.getQk());
  606. callback.setTagId(task.tagId);
  607. callback.setCallbackUrl(record.getCallbackUrl());
  608. callback.setEventType(record.getEventType());
  609. callback.setEventTimeMs(record.getEventTimeMs());
  610. callback.setPurchase(record.getPurchase());
  611. callback.setStatus(record.getStatus());
  612. callback.setOk(record.isOk());
  613. callback.setAttempt(record.getAttempt());
  614. callback.setResponseBody(record.getResponseBody());
  615. callback.setErrorMessage(record.getErrorMessage());
  616. callback.setRequestBody(record.getRequestBody());
  617. callback.setCreatedAt(record.getCreatedAt());
  618. return callback;
  619. }
  620. private static MediaCallbackRecord toCallbackRecord(CallbackTask task, HonorMediaCallbackRecord honorRecord) {
  621. MediaCallbackRecord callbackRecord = new MediaCallbackRecord();
  622. callbackRecord.setDedupeKey(task.callbackKey);
  623. callbackRecord.setMedia("honor");
  624. callbackRecord.setQk(honorRecord.getQk());
  625. callbackRecord.setCallbackUrl(honorRecord.getCallbackUrl());
  626. callbackRecord.setEventType(honorRecord.getEventType());
  627. callbackRecord.setEventTimeMs(honorRecord.getEventTimeMs());
  628. callbackRecord.setPurchase(honorRecord.getPurchase());
  629. callbackRecord.setStatus(honorRecord.getStatus());
  630. callbackRecord.setOk(honorRecord.isOk());
  631. callbackRecord.setAttempt(honorRecord.getAttempt());
  632. callbackRecord.setResponseBody(honorRecord.getResponseBody());
  633. callbackRecord.setErrorMessage(honorRecord.getErrorMessage());
  634. callbackRecord.setRequestBody(honorRecord.getRequestBody());
  635. callbackRecord.setTrackingVersion("v1");
  636. callbackRecord.setCreatedAt(honorRecord.getCreatedAt());
  637. return callbackRecord;
  638. }
  639. // ─── 内部数据结构 ────────────────────────────────────────────────────────
  640. private static class BidLookup {
  641. private final Map<String, BidRecord> tencentBids;
  642. private final Map<String, HonorBidRecord> honorBids;
  643. private BidLookup(Map<String, BidRecord> tencentBids, Map<String, HonorBidRecord> honorBids) {
  644. this.tencentBids = tencentBids;
  645. this.honorBids = honorBids;
  646. }
  647. }
  648. private static class CallbackTask {
  649. private final String media;
  650. private final PaymentInfo payment;
  651. private final String tagId;
  652. private final String callbackKey;
  653. private final BidRecord tencentBid;
  654. private final String tencentActionType;
  655. private final HonorBidRecord honorBid;
  656. private final Integer honorConversionId;
  657. private CallbackTask(String media,
  658. PaymentInfo payment,
  659. String tagId,
  660. String callbackKey,
  661. BidRecord tencentBid,
  662. String tencentActionType,
  663. HonorBidRecord honorBid,
  664. Integer honorConversionId) {
  665. this.media = media;
  666. this.payment = payment;
  667. this.tagId = tagId;
  668. this.callbackKey = callbackKey;
  669. this.tencentBid = tencentBid;
  670. this.tencentActionType = tencentActionType;
  671. this.honorBid = honorBid;
  672. this.honorConversionId = honorConversionId;
  673. }
  674. private static CallbackTask tencent(PaymentInfo payment,
  675. BidRecord bid,
  676. String callbackKey,
  677. String actionType) {
  678. return new CallbackTask("tencent", payment, bid.getTagId(), callbackKey, bid, actionType, null, null);
  679. }
  680. private static CallbackTask honor(PaymentInfo payment,
  681. HonorBidRecord bid,
  682. String callbackKey,
  683. Integer conversionId) {
  684. return new CallbackTask("honor", payment, bid.getTagId(), callbackKey, null, null, bid, conversionId);
  685. }
  686. private boolean isTencent() {
  687. return "tencent".equals(media);
  688. }
  689. private boolean isHonor() {
  690. return "honor".equals(media);
  691. }
  692. }
  693. private static class CallbackDispatchResult {
  694. private final String media;
  695. private final CallbackTask task;
  696. private final MediaCallbackRecord record;
  697. private CallbackDispatchResult(String media, CallbackTask task, MediaCallbackRecord record) {
  698. this.media = media;
  699. this.task = task;
  700. this.record = record;
  701. }
  702. private static CallbackDispatchResult tencent(CallbackTask task, MediaCallbackRecord record) {
  703. return new CallbackDispatchResult("tencent", task, record);
  704. }
  705. private static CallbackDispatchResult honor(CallbackTask task, MediaCallbackRecord record) {
  706. return new CallbackDispatchResult("honor", task, record);
  707. }
  708. private boolean isHonor() {
  709. return "honor".equals(media);
  710. }
  711. }
  712. public static class SyncResult {
  713. public final AtomicInteger fetched = new AtomicInteger();
  714. public final AtomicInteger matched = new AtomicInteger();
  715. public final AtomicInteger sent = new AtomicInteger();
  716. public final AtomicInteger skipped = new AtomicInteger();
  717. public final AtomicInteger failed = new AtomicInteger();
  718. public final AtomicInteger alreadySent = new AtomicInteger();
  719. }
  720. }