TiDBColdStore.java 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546
  1. package com.adx.tencent.storage;
  2. import com.adx.tencent.storage.mapper.*;
  3. import com.adx.tencent.storage.model.*;
  4. import com.fasterxml.jackson.databind.ObjectMapper;
  5. import org.apache.ibatis.session.ExecutorType;
  6. import org.apache.ibatis.session.SqlSession;
  7. import org.apache.ibatis.session.SqlSessionFactory;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10. import java.sql.Statement;
  11. import java.sql.Timestamp;
  12. import java.time.Instant;
  13. import java.util.List;
  14. /**
  15. * 对应 Go internal/storage/tidb.go 的 TiDBColdStore。
  16. * 使用 MyBatis 操作 TiDB(MySQL 协议兼容)。
  17. * 功能:
  18. * 1. migrate() - 建表 DDL
  19. * 2. saveBid / saveTracking / saveConversion / saveMediaCallback
  20. * 3. successfulCallbackExists / successfulCallbackExistsForEvent
  21. * 4. pendingMediaCallbacks
  22. */
  23. public class TiDBColdStore {
  24. private static final Logger log = LoggerFactory.getLogger(TiDBColdStore.class);
  25. private static final int CONVERSION_BATCH_SIZE = 1000;
  26. private final SqlSessionFactory sqlSessionFactory;
  27. private final ObjectMapper objectMapper;
  28. public TiDBColdStore(SqlSessionFactory sqlSessionFactory, ObjectMapper objectMapper) {
  29. this.sqlSessionFactory = sqlSessionFactory;
  30. this.objectMapper = objectMapper;
  31. }
  32. // ─── DDL 迁移 ────────────────────────────────────────────────────────────
  33. public void migrate() {
  34. String[] ddls = {
  35. """
  36. CREATE TABLE IF NOT EXISTS tencent_ad_bid_events (
  37. id BIGINT AUTO_INCREMENT PRIMARY KEY,
  38. qk VARCHAR(128) NOT NULL,
  39. media VARCHAR(64),
  40. media_trace_id VARCHAR(255),
  41. platform VARCHAR(16),
  42. req_id VARCHAR(512),
  43. bid_id VARCHAR(128),
  44. creative_id VARCHAR(128),
  45. imp_id VARCHAR(128),
  46. tag_id VARCHAR(128),
  47. price BIGINT UNSIGNED,
  48. show_urls JSON,
  49. click_urls JSON,
  50. landing_page TEXT,
  51. app_store_link TEXT,
  52. package_name VARCHAR(255),
  53. media_params JSON,
  54. created_at TIMESTAMP(3) NOT NULL,
  55. UNIQUE KEY uk_tencent_ad_bid_events_qk (qk),
  56. KEY idx_tencent_ad_bid_events_media_trace (media, media_trace_id),
  57. KEY idx_tencent_ad_bid_events_platform_tag (platform, tag_id),
  58. KEY idx_tencent_ad_bid_events_created_at (created_at)
  59. )
  60. """,
  61. """
  62. CREATE TABLE IF NOT EXISTS tencent_tracking_reports (
  63. id BIGINT AUTO_INCREMENT PRIMARY KEY,
  64. qk VARCHAR(128),
  65. kind VARCHAR(32) NOT NULL,
  66. tracking_url TEXT NOT NULL,
  67. status INT NOT NULL,
  68. ok TINYINT(1) NOT NULL,
  69. created_at TIMESTAMP(3) NOT NULL,
  70. KEY idx_tencent_tracking_reports_qk_kind_created_at (qk, kind, created_at)
  71. )
  72. """,
  73. """
  74. CREATE TABLE IF NOT EXISTS tencent_baidu_conversions (
  75. id BIGINT AUTO_INCREMENT PRIMARY KEY,
  76. dedupe_key VARCHAR(255),
  77. qk VARCHAR(128),
  78. media VARCHAR(64),
  79. tag_id VARCHAR(128),
  80. date_value VARCHAR(16),
  81. appsid VARCHAR(128),
  82. customer_name VARCHAR(128),
  83. device_id VARCHAR(255),
  84. conv DOUBLE,
  85. payment DOUBLE,
  86. gmv DOUBLE,
  87. act INT,
  88. tu VARCHAR(128),
  89. clk_time VARCHAR(32),
  90. created_at TIMESTAMP(3) NOT NULL,
  91. UNIQUE KEY uk_tencent_baidu_conversions_dedupe (dedupe_key),
  92. KEY idx_tencent_baidu_conversions_qk_act_created_at (qk, act, created_at),
  93. KEY idx_tencent_baidu_conversions_tag_id (tag_id),
  94. KEY idx_tencent_baidu_conversions_device_id (device_id)
  95. )
  96. """,
  97. """
  98. CREATE TABLE IF NOT EXISTS tencent_media_callbacks (
  99. id BIGINT AUTO_INCREMENT PRIMARY KEY,
  100. dedupe_key VARCHAR(255),
  101. media VARCHAR(64) NOT NULL,
  102. qk VARCHAR(128),
  103. callback_url TEXT NOT NULL,
  104. event_type INT NOT NULL,
  105. event_time_ms BIGINT NOT NULL,
  106. purchase DOUBLE,
  107. status INT NOT NULL,
  108. ok TINYINT(1) NOT NULL,
  109. attempt INT NOT NULL DEFAULT 1,
  110. response_body TEXT,
  111. error_message TEXT,
  112. request_body TEXT,
  113. dispatch_status VARCHAR(32) NOT NULL DEFAULT 'SENT',
  114. tracking_version VARCHAR(16) NOT NULL DEFAULT '0',
  115. created_at TIMESTAMP(3) NOT NULL,
  116. KEY idx_tencent_media_callbacks_dedupe_ok_created_at (dedupe_key, ok, created_at),
  117. KEY idx_tencent_media_callbacks_qk_created_at (qk, created_at),
  118. KEY idx_tencent_media_callbacks_media_event_created_at (media, event_type, created_at)
  119. )
  120. """,
  121. """
  122. CREATE TABLE IF NOT EXISTS tencent_tag_event (
  123. tag_id VARCHAR(32) NOT NULL,
  124. baidu_act INT NOT NULL COMMENT '百度转化行为',
  125. tencent_action_type VARCHAR(128) NOT NULL COMMENT '腾讯回传行为',
  126. PRIMARY KEY (tag_id, baidu_act)
  127. ) COMMENT '竞价事件记录'
  128. """,
  129. """
  130. CREATE TABLE IF NOT EXISTS tencent_account_tag_event (
  131. tag_id VARCHAR(128) NOT NULL COMMENT '百度广告位ID',
  132. account_id VARCHAR(64) NOT NULL COMMENT '腾讯广告主ID',
  133. baidu_act INT NOT NULL COMMENT '百度转化行为',
  134. tencent_action_type VARCHAR(128) NOT NULL COMMENT '腾讯回传行为',
  135. tencent_custom_action VARCHAR(128) COMMENT '腾讯自定义行为(action_type=CUSTOM时使用)',
  136. created_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间',
  137. updated_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '更新时间',
  138. PRIMARY KEY (tag_id, account_id, baidu_act),
  139. KEY idx_tencent_account_tag_event_account_act (account_id, baidu_act)
  140. ) COMMENT '腾讯账户级广告位回传方式配置'
  141. """,
  142. """
  143. CREATE TABLE IF NOT EXISTS tencent_tag_event_profile (
  144. tag_id VARCHAR(128) NOT NULL COMMENT '百度广告位ID',
  145. callback_profile VARCHAR(64) NOT NULL COMMENT '回传方案',
  146. baidu_act INT NOT NULL COMMENT '百度转化行为',
  147. tencent_action_type VARCHAR(128) NOT NULL COMMENT '腾讯回传行为',
  148. tencent_custom_action VARCHAR(128) COMMENT '腾讯自定义行为(action_type=CUSTOM时使用)',
  149. enabled TINYINT(1) NOT NULL DEFAULT 1 COMMENT '是否启用',
  150. created_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间',
  151. updated_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '更新时间',
  152. PRIMARY KEY (tag_id, callback_profile, baidu_act),
  153. KEY idx_tencent_tag_event_profile_profile_act (callback_profile, baidu_act)
  154. ) COMMENT '腾讯profile级广告位回传方式配置'
  155. """
  156. };
  157. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  158. Statement stmt = session.getConnection().createStatement();
  159. for (String ddl : ddls) {
  160. stmt.execute(ddl);
  161. }
  162. // 字段长度补丁(已存在的表可能是旧的 128)
  163. try {
  164. stmt.execute("ALTER TABLE tencent_ad_bid_events MODIFY COLUMN req_id VARCHAR(512)");
  165. } catch (Exception ignored) {}
  166. // platform 字段补丁
  167. try {
  168. stmt.execute("ALTER TABLE tencent_ad_bid_events ADD COLUMN platform VARCHAR(16) AFTER media_trace_id");
  169. } catch (Exception ignored) {}
  170. try {
  171. stmt.execute("ALTER TABLE tencent_ad_bid_events ADD KEY idx_tencent_ad_bid_events_platform_tag (platform, tag_id)");
  172. } catch (Exception ignored) {}
  173. try {
  174. stmt.execute("ALTER TABLE tencent_baidu_conversions ADD COLUMN gmv DOUBLE AFTER payment");
  175. } catch (Exception ignored) {}
  176. try {
  177. stmt.execute("ALTER TABLE tencent_baidu_conversions ADD COLUMN tag_id VARCHAR(128) AFTER media");
  178. } catch (Exception ignored) {}
  179. try {
  180. stmt.execute("ALTER TABLE tencent_baidu_conversions ADD KEY idx_tencent_baidu_conversions_tag_id (tag_id)");
  181. } catch (Exception ignored) {}
  182. try {
  183. stmt.execute("ALTER TABLE tencent_media_callbacks ADD COLUMN request_body TEXT AFTER error_message");
  184. } catch (Exception ignored) {}
  185. try {
  186. stmt.execute("ALTER TABLE tencent_media_callbacks ADD COLUMN dispatch_status VARCHAR(32) NOT NULL DEFAULT 'SENT' AFTER request_body");
  187. } catch (Exception ignored) {}
  188. try {
  189. stmt.execute("ALTER TABLE tencent_tracking_reports ADD KEY idx_tencent_tracking_reports_created_at_id (created_at, id)");
  190. } catch (Exception ignored) {}
  191. try {
  192. stmt.execute("UPDATE tencent_media_callbacks SET dispatch_status = 'SENT' WHERE dispatch_status IS NULL OR dispatch_status = ''");
  193. } catch (Exception ignored) {}
  194. try {
  195. stmt.execute("ALTER TABLE tencent_media_callbacks ADD COLUMN tracking_version VARCHAR(16) NOT NULL DEFAULT '0' AFTER dispatch_status");
  196. } catch (Exception ignored) {}
  197. try {
  198. stmt.execute("UPDATE tencent_media_callbacks SET tracking_version = '0' WHERE tracking_version IS NULL OR tracking_version = ''");
  199. } catch (Exception ignored) {}
  200. try {
  201. stmt.execute("ALTER TABLE tencent_tag_event ADD COLUMN baidu_act INT NULL COMMENT '百度转化行为' AFTER tag_id");
  202. } catch (Exception ignored) {}
  203. try {
  204. stmt.execute("ALTER TABLE tencent_tag_event CHANGE COLUMN event_type tencent_action_type VARCHAR(128) NOT NULL COMMENT '腾讯回传行为'");
  205. } catch (Exception ignored) {}
  206. try {
  207. stmt.execute("UPDATE tencent_tag_event SET baidu_act = 0 WHERE baidu_act IS NULL");
  208. } catch (Exception ignored) {}
  209. try {
  210. stmt.execute("ALTER TABLE tencent_tag_event MODIFY COLUMN baidu_act INT NOT NULL");
  211. } catch (Exception ignored) {}
  212. try {
  213. stmt.execute("ALTER TABLE tencent_tag_event DROP PRIMARY KEY");
  214. } catch (Exception ignored) {}
  215. try {
  216. stmt.execute("ALTER TABLE tencent_tag_event ADD PRIMARY KEY (tag_id, baidu_act)");
  217. } catch (Exception ignored) {}
  218. try {
  219. stmt.execute("ALTER TABLE tencent_account_tag_event ADD COLUMN tencent_custom_action VARCHAR(128) COMMENT '腾讯自定义行为(action_type=CUSTOM时使用)' AFTER tencent_action_type");
  220. } catch (Exception ignored) {}
  221. try {
  222. stmt.execute("""
  223. CREATE TABLE IF NOT EXISTS tencent_tag_event_profile (
  224. tag_id VARCHAR(128) NOT NULL COMMENT '百度广告位ID',
  225. callback_profile VARCHAR(64) NOT NULL COMMENT '回传方案',
  226. baidu_act INT NOT NULL COMMENT '百度转化行为',
  227. tencent_action_type VARCHAR(128) NOT NULL COMMENT '腾讯回传行为',
  228. tencent_custom_action VARCHAR(128) COMMENT '腾讯自定义行为(action_type=CUSTOM时使用)',
  229. enabled TINYINT(1) NOT NULL DEFAULT 1 COMMENT '是否启用',
  230. created_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间',
  231. updated_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '更新时间',
  232. PRIMARY KEY (tag_id, callback_profile, baidu_act),
  233. KEY idx_tencent_tag_event_profile_profile_act (callback_profile, baidu_act)
  234. ) COMMENT '腾讯profile级广告位回传方式配置'
  235. """);
  236. } catch (Exception ignored) {}
  237. stmt.close();
  238. } catch (Exception e) {
  239. throw new RuntimeException("migrate failed", e);
  240. }
  241. }
  242. // ─── SaveBid ─────────────────────────────────────────────────────────────
  243. public void saveBid(BidRecord record) {
  244. if (record.getQk() == null || record.getQk().isBlank()) {
  245. throw new IllegalArgumentException("qk is required");
  246. }
  247. if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
  248. record.setMediaTraceId(record.getMediaTraceId());
  249. BidRecord cold = record.toColdRecord();
  250. String mediaParams = toJson(cold.getMediaParams());
  251. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  252. BidEventMapper mapper = session.getMapper(BidEventMapper.class);
  253. mapper.insertOrUpdate(
  254. cold.getQk(), cold.getMedia(), cold.getMediaTraceId(),
  255. cold.getPlatform(), cold.getAccountId(), cold.getEventType(),
  256. cold.getTagId(), cold.getPrice() != null ? cold.getPrice() : 0L,
  257. cold.getPackageName(), mediaParams, Timestamp.from(cold.getCreatedAt())
  258. );
  259. }
  260. }
  261. public BidRecord getBidByQk(String qk) {
  262. if (qk == null || qk.isBlank()) return null;
  263. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  264. BidEventMapper mapper = session.getMapper(BidEventMapper.class);
  265. return mapper.selectByQk(qk);
  266. }
  267. }
  268. public List<BidRecord> getBidsByQks(List<String> qks) {
  269. if (qks == null || qks.isEmpty()) return List.of();
  270. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  271. BidEventMapper mapper = session.getMapper(BidEventMapper.class);
  272. return mapper.selectByQks(qks);
  273. }
  274. }
  275. // ─── SaveTracking ─────────────────────────────────────────────────────────
  276. public void saveTracking(TrackingRecord record) {
  277. if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
  278. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  279. TrackingReportMapper mapper = session.getMapper(TrackingReportMapper.class);
  280. mapper.insert(
  281. record.getQk(), record.getKind(), record.getUrl(),
  282. record.getStatus(), record.isOk() ? 1 : 0,
  283. Timestamp.from(record.getCreatedAt())
  284. );
  285. }
  286. }
  287. // ─── SaveConversion ──────────────────────────────────────────────────────
  288. public void saveConversion(ConversionRecord record) {
  289. if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
  290. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  291. ConversionMapper mapper = session.getMapper(ConversionMapper.class);
  292. mapper.insertOrUpdate(
  293. record.getDedupeKey(), record.getQk(), record.getMedia(), record.getTagId(),
  294. record.getDate(), record.getAppSid(), record.getCustomerName(),
  295. record.getDeviceId(), record.getConv(), record.getPayment(), record.getGmv(),
  296. record.getAct(), record.getTu(), record.getClkTime(),
  297. Timestamp.from(record.getCreatedAt())
  298. );
  299. }
  300. }
  301. public void saveConversions(List<ConversionRecord> records) {
  302. if (records == null || records.isEmpty()) return;
  303. long startedAt = System.currentTimeMillis();
  304. Instant now = Instant.now();
  305. for (ConversionRecord record : records) {
  306. if (record.getCreatedAt() == null) record.setCreatedAt(now);
  307. }
  308. try (SqlSession session = sqlSessionFactory.openSession(false)) {
  309. ConversionMapper mapper = session.getMapper(ConversionMapper.class);
  310. for (int start = 0; start < records.size(); start += CONVERSION_BATCH_SIZE) {
  311. int end = Math.min(start + CONVERSION_BATCH_SIZE, records.size());
  312. mapper.batchInsertOrUpdate(records.subList(start, end));
  313. }
  314. session.commit();
  315. long costMs = System.currentTimeMillis() - startedAt;
  316. int batchCount = (records.size() + CONVERSION_BATCH_SIZE - 1) / CONVERSION_BATCH_SIZE;
  317. log.info("[TiDBColdStore] saveConversions done: size={}, batches={}, batchSize={}, cost={}ms",
  318. records.size(), batchCount, CONVERSION_BATCH_SIZE, costMs);
  319. }
  320. }
  321. // ─── SaveMediaCallback ───────────────────────────────────────────────────
  322. public void saveMediaCallback(MediaCallbackRecord record) {
  323. if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
  324. int attempt = record.getAttempt() <= 0 ? 1 : record.getAttempt();
  325. if (record.getDispatchStatus() == null || record.getDispatchStatus().isBlank()) {
  326. record.setDispatchStatus(MediaCallbackRecord.DISPATCH_STATUS_SENT);
  327. }
  328. if (record.getTrackingVersion() == null || record.getTrackingVersion().isBlank()) {
  329. record.setTrackingVersion(defaultTrackingVersion(record.getMedia()));
  330. }
  331. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  332. MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
  333. mapper.insert(
  334. record.getDedupeKey(), record.getMedia(), record.getQk(),
  335. record.getCallbackUrl(), record.getEventType(), record.getEventTimeMs(),
  336. record.getPurchase(), record.getStatus(), record.isOk() ? 1 : 0,
  337. attempt, record.getResponseBody(), record.getErrorMessage(),
  338. record.getRequestBody(), record.getDispatchStatus(), record.getTrackingVersion(),
  339. Timestamp.from(record.getCreatedAt())
  340. );
  341. }
  342. }
  343. public void saveMediaCallbacks(List<MediaCallbackRecord> records) {
  344. if (records == null || records.isEmpty()) return;
  345. for (MediaCallbackRecord record : records) {
  346. if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
  347. if (record.getAttempt() <= 0) record.setAttempt(1);
  348. if (record.getDispatchStatus() == null || record.getDispatchStatus().isBlank()) {
  349. record.setDispatchStatus(MediaCallbackRecord.DISPATCH_STATUS_SENT);
  350. }
  351. if (record.getTrackingVersion() == null || record.getTrackingVersion().isBlank()) {
  352. record.setTrackingVersion(defaultTrackingVersion(record.getMedia()));
  353. }
  354. }
  355. try (SqlSession session = sqlSessionFactory.openSession(ExecutorType.BATCH, false)) {
  356. MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
  357. mapper.batchInsert(records);
  358. session.commit();
  359. }
  360. }
  361. // ─── 查询:成功回调是否存在 ────────────────────────────────────────────────
  362. public boolean successfulCallbackExists(String dedupeKey) {
  363. if (dedupeKey == null || dedupeKey.isBlank()) return false;
  364. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  365. MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
  366. int count = mapper.countSuccessfulByDedupeKey(dedupeKey);
  367. return count > 0;
  368. }
  369. }
  370. public boolean successfulCallbackExistsForEvent(String dedupeKey, String legacyDedupeKey, int eventType) {
  371. if (dedupeKey == null || dedupeKey.isBlank()) return false;
  372. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  373. MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
  374. int count = mapper.countSuccessfulForEvent(dedupeKey, legacyDedupeKey, eventType);
  375. return count > 0;
  376. }
  377. }
  378. public List<String> successfulCallbackKeys(List<String> dedupeKeys) {
  379. if (dedupeKeys == null || dedupeKeys.isEmpty()) return List.of();
  380. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  381. MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
  382. return mapper.selectSuccessfulDedupeKeys(dedupeKeys);
  383. }
  384. }
  385. public List<String> successfulCallbackKeysWithLegacy(List<String> dedupeKeys,
  386. List<String> legacyDedupeKeys,
  387. List<Integer> eventTypes) {
  388. if (dedupeKeys == null || dedupeKeys.isEmpty()) return List.of();
  389. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  390. MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
  391. return mapper.selectSuccessfulKeysWithLegacy(dedupeKeys, legacyDedupeKeys, eventTypes);
  392. }
  393. }
  394. // ─── 查询:待重试的回调 ────────────────────────────────────────────────────
  395. public List<MediaCallbackRecord> pendingMediaCallbacks(String media, int limit) {
  396. if (limit <= 0) limit = 100;
  397. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  398. MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
  399. return mapper.selectPendingCallbacks(media, limit);
  400. }
  401. }
  402. public MediaCallbackRecord getMediaCallbackById(long id) {
  403. if (id <= 0) return null;
  404. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  405. MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
  406. return mapper.selectById(id);
  407. }
  408. }
  409. public void updateMediaCallback(MediaCallbackRecord record) {
  410. if (record == null || record.getId() == null || record.getId() <= 0) {
  411. throw new IllegalArgumentException("media callback id is required");
  412. }
  413. int attempt = record.getAttempt() <= 0 ? 1 : record.getAttempt();
  414. record.setAttempt(attempt);
  415. if (record.getDispatchStatus() == null || record.getDispatchStatus().isBlank()) {
  416. record.setDispatchStatus(MediaCallbackRecord.DISPATCH_STATUS_SENT);
  417. }
  418. if (record.getTrackingVersion() == null || record.getTrackingVersion().isBlank()) {
  419. record.setTrackingVersion(defaultTrackingVersion(record.getMedia()));
  420. }
  421. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  422. MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
  423. mapper.updateById(record);
  424. }
  425. }
  426. private static String defaultTrackingVersion(String media) {
  427. return "tencent".equalsIgnoreCase(media) ? "0" : "v1";
  428. }
  429. // ─── 查询:广告位回传方式配置 ──────────────────────────────────────────────
  430. public List<TagEventRecord> listAllTagEvents() {
  431. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  432. TagEventMapper mapper = session.getMapper(TagEventMapper.class);
  433. return mapper.selectAll();
  434. }
  435. }
  436. /**
  437. * 根据广告位ID查询回传方式配置(DB 兜底查询)。
  438. */
  439. public TagEventRecord getTagEvent(String tagId, int baiduAct) {
  440. if (tagId == null || tagId.isBlank()) return null;
  441. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  442. TagEventMapper mapper = session.getMapper(TagEventMapper.class);
  443. return mapper.selectByTagIdAndAct(tagId, baiduAct);
  444. }
  445. }
  446. public TencentTagEventProfileRecord getTagEventProfile(String tagId, String callbackProfile, int baiduAct) {
  447. if (tagId == null || tagId.isBlank() || callbackProfile == null || callbackProfile.isBlank()) {
  448. return null;
  449. }
  450. String sql = """
  451. SELECT tag_id, callback_profile, baidu_act, tencent_action_type, tencent_custom_action
  452. FROM tencent_tag_event_profile
  453. WHERE tag_id = ? AND callback_profile = ? AND baidu_act = ? AND enabled = 1
  454. """;
  455. try (SqlSession session = sqlSessionFactory.openSession(true);
  456. var ps = session.getConnection().prepareStatement(sql)) {
  457. ps.setString(1, tagId);
  458. ps.setString(2, callbackProfile);
  459. ps.setInt(3, baiduAct);
  460. try (var rs = ps.executeQuery()) {
  461. if (!rs.next()) {
  462. return null;
  463. }
  464. TencentTagEventProfileRecord record = new TencentTagEventProfileRecord();
  465. record.setTagId(rs.getString(1));
  466. record.setCallbackProfile(rs.getString(2));
  467. record.setBaiduAct(rs.getInt(3));
  468. record.setTencentActionType(rs.getString(4));
  469. record.setTencentCustomAction(rs.getString(5));
  470. return record;
  471. }
  472. } catch (Exception e) {
  473. log.warn("[TiDBColdStore] 查询 tencent_tag_event_profile 失败 | tagId={} | profile={} | act={} | error={}",
  474. tagId, callbackProfile, baiduAct, e.getMessage());
  475. return null;
  476. }
  477. }
  478. // ─── 查询:腾讯账户级广告位回传方式配置 ───────────────────────────────────
  479. public List<AccountTagEventRecord> listAllAccountTagEvents() {
  480. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  481. AccountTagEventMapper mapper = session.getMapper(AccountTagEventMapper.class);
  482. return mapper.selectAll();
  483. }
  484. }
  485. /**
  486. * 根据广告位ID、腾讯广告主ID和百度行为查询账户级回传方式配置(DB 兜底查询)。
  487. */
  488. public AccountTagEventRecord getAccountTagEvent(String tagId, String accountId, int baiduAct) {
  489. if (tagId == null || tagId.isBlank()) return null;
  490. if (accountId == null || accountId.isBlank()) return null;
  491. try (SqlSession session = sqlSessionFactory.openSession(true)) {
  492. AccountTagEventMapper mapper = session.getMapper(AccountTagEventMapper.class);
  493. return mapper.selectByTagIdAndAccountIdAndAct(tagId, accountId, baiduAct);
  494. }
  495. }
  496. // ─── 工具方法 ─────────────────────────────────────────────────────────────
  497. private String toJson(Object value) {
  498. if (value == null) return "null";
  499. try {
  500. return objectMapper.writeValueAsString(value);
  501. } catch (Exception e) {
  502. throw new RuntimeException("JSON serialize failed", e);
  503. }
  504. }
  505. }