|
@@ -1,18 +1,19 @@
|
|
|
package com.adx.tracking.storage;
|
|
package com.adx.tracking.storage;
|
|
|
|
|
|
|
|
|
|
+import com.adx.tracking.storage.mapper.*;
|
|
|
import com.adx.tracking.storage.model.*;
|
|
import com.adx.tracking.storage.model.*;
|
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
-import org.springframework.jdbc.core.JdbcTemplate;
|
|
|
|
|
|
|
+import org.apache.ibatis.session.SqlSession;
|
|
|
|
|
+import org.apache.ibatis.session.SqlSessionFactory;
|
|
|
|
|
|
|
|
-import javax.sql.DataSource;
|
|
|
|
|
|
|
+import java.sql.Statement;
|
|
|
import java.sql.Timestamp;
|
|
import java.sql.Timestamp;
|
|
|
import java.time.Instant;
|
|
import java.time.Instant;
|
|
|
-import java.util.ArrayList;
|
|
|
|
|
import java.util.List;
|
|
import java.util.List;
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
* 对应 Go internal/storage/tidb.go 的 TiDBColdStore。
|
|
* 对应 Go internal/storage/tidb.go 的 TiDBColdStore。
|
|
|
- * 使用 Spring JdbcTemplate 操作 TiDB(MySQL 协议兼容)。
|
|
|
|
|
|
|
+ * 使用 MyBatis 操作 TiDB(MySQL 协议兼容)。
|
|
|
* 功能:
|
|
* 功能:
|
|
|
* 1. migrate() - 建表 DDL
|
|
* 1. migrate() - 建表 DDL
|
|
|
* 2. saveBid / saveTracking / saveConversion / saveMediaCallback
|
|
* 2. saveBid / saveTracking / saveConversion / saveMediaCallback
|
|
@@ -21,11 +22,11 @@ import java.util.List;
|
|
|
*/
|
|
*/
|
|
|
public class TiDBColdStore {
|
|
public class TiDBColdStore {
|
|
|
|
|
|
|
|
- private final JdbcTemplate jdbc;
|
|
|
|
|
|
|
+ private final SqlSessionFactory sqlSessionFactory;
|
|
|
private final ObjectMapper objectMapper;
|
|
private final ObjectMapper objectMapper;
|
|
|
|
|
|
|
|
- public TiDBColdStore(DataSource dataSource, ObjectMapper objectMapper) {
|
|
|
|
|
- this.jdbc = new JdbcTemplate(dataSource);
|
|
|
|
|
|
|
+ public TiDBColdStore(SqlSessionFactory sqlSessionFactory, ObjectMapper objectMapper) {
|
|
|
|
|
+ this.sqlSessionFactory = sqlSessionFactory;
|
|
|
this.objectMapper = objectMapper;
|
|
this.objectMapper = objectMapper;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -112,13 +113,19 @@ public class TiDBColdStore {
|
|
|
)
|
|
)
|
|
|
"""
|
|
"""
|
|
|
};
|
|
};
|
|
|
- for (String ddl : ddls) {
|
|
|
|
|
- jdbc.execute(ddl);
|
|
|
|
|
|
|
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
|
|
|
|
|
+ Statement stmt = session.getConnection().createStatement();
|
|
|
|
|
+ for (String ddl : ddls) {
|
|
|
|
|
+ stmt.execute(ddl);
|
|
|
|
|
+ }
|
|
|
|
|
+ // 字段长度补丁(已存在的表可能是旧的 128)
|
|
|
|
|
+ try {
|
|
|
|
|
+ stmt.execute("ALTER TABLE ad_bid_events MODIFY COLUMN req_id VARCHAR(512)");
|
|
|
|
|
+ } catch (Exception ignored) {}
|
|
|
|
|
+ stmt.close();
|
|
|
|
|
+ } catch (Exception e) {
|
|
|
|
|
+ throw new RuntimeException("migrate failed", e);
|
|
|
}
|
|
}
|
|
|
- // 字段长度补丁(已存在的表可能是旧的 128)
|
|
|
|
|
- try {
|
|
|
|
|
- jdbc.execute("ALTER TABLE ad_bid_events MODIFY COLUMN req_id VARCHAR(512)");
|
|
|
|
|
- } catch (Exception ignored) {}
|
|
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// ─── SaveBid ─────────────────────────────────────────────────────────────
|
|
// ─── SaveBid ─────────────────────────────────────────────────────────────
|
|
@@ -133,72 +140,47 @@ public class TiDBColdStore {
|
|
|
String clickUrls = toJson(record.getClickUrls());
|
|
String clickUrls = toJson(record.getClickUrls());
|
|
|
String mediaParams = toJson(record.getMediaParams());
|
|
String mediaParams = toJson(record.getMediaParams());
|
|
|
|
|
|
|
|
- jdbc.update("""
|
|
|
|
|
- INSERT INTO ad_bid_events (
|
|
|
|
|
- qk, media, media_trace_id, req_id, bid_id, creative_id, imp_id, tag_id, price,
|
|
|
|
|
- show_urls, click_urls, landing_page, app_store_link, package_name,
|
|
|
|
|
- media_params, created_at
|
|
|
|
|
- ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
|
|
|
- ON DUPLICATE KEY UPDATE
|
|
|
|
|
- media = VALUES(media),
|
|
|
|
|
- media_trace_id = VALUES(media_trace_id),
|
|
|
|
|
- req_id = VALUES(req_id),
|
|
|
|
|
- bid_id = VALUES(bid_id),
|
|
|
|
|
- creative_id = VALUES(creative_id),
|
|
|
|
|
- imp_id = VALUES(imp_id),
|
|
|
|
|
- tag_id = VALUES(tag_id),
|
|
|
|
|
- price = VALUES(price),
|
|
|
|
|
- show_urls = VALUES(show_urls),
|
|
|
|
|
- click_urls = VALUES(click_urls),
|
|
|
|
|
- landing_page = VALUES(landing_page),
|
|
|
|
|
- app_store_link = VALUES(app_store_link),
|
|
|
|
|
- package_name = VALUES(package_name),
|
|
|
|
|
- media_params = VALUES(media_params)
|
|
|
|
|
- """,
|
|
|
|
|
- record.getQk(), record.getMedia(), record.getMediaTraceId(),
|
|
|
|
|
- record.getReqId(), record.getBidId(), record.getCreativeId(),
|
|
|
|
|
- record.getImpId(), record.getTagId(), record.getPrice(),
|
|
|
|
|
- showUrls, clickUrls, record.getLandingPage(),
|
|
|
|
|
- record.getAppStoreLink(), record.getPackageName(),
|
|
|
|
|
- mediaParams, Timestamp.from(record.getCreatedAt())
|
|
|
|
|
- );
|
|
|
|
|
|
|
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
|
|
|
|
|
+ BidEventMapper mapper = session.getMapper(BidEventMapper.class);
|
|
|
|
|
+ mapper.insertOrUpdate(
|
|
|
|
|
+ record.getQk(), record.getMedia(), record.getMediaTraceId(),
|
|
|
|
|
+ record.getReqId(), record.getBidId(), record.getCreativeId(),
|
|
|
|
|
+ record.getImpId(), record.getTagId(), record.getPrice(),
|
|
|
|
|
+ showUrls, clickUrls, record.getLandingPage(),
|
|
|
|
|
+ record.getAppStoreLink(), record.getPackageName(),
|
|
|
|
|
+ mediaParams, Timestamp.from(record.getCreatedAt())
|
|
|
|
|
+ );
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// ─── SaveTracking ─────────────────────────────────────────────────────────
|
|
// ─── SaveTracking ─────────────────────────────────────────────────────────
|
|
|
|
|
|
|
|
public void saveTracking(TrackingRecord record) {
|
|
public void saveTracking(TrackingRecord record) {
|
|
|
if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
|
|
if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
|
|
|
- jdbc.update("""
|
|
|
|
|
- INSERT INTO tracking_reports (qk, kind, tracking_url, status, ok, created_at)
|
|
|
|
|
- VALUES (?, ?, ?, ?, ?, ?)
|
|
|
|
|
- """,
|
|
|
|
|
- record.getQk(), record.getKind(), record.getUrl(),
|
|
|
|
|
- record.getStatus(), record.isOk() ? 1 : 0,
|
|
|
|
|
- Timestamp.from(record.getCreatedAt())
|
|
|
|
|
- );
|
|
|
|
|
|
|
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
|
|
|
|
|
+ TrackingReportMapper mapper = session.getMapper(TrackingReportMapper.class);
|
|
|
|
|
+ mapper.insert(
|
|
|
|
|
+ record.getQk(), record.getKind(), record.getUrl(),
|
|
|
|
|
+ record.getStatus(), record.isOk() ? 1 : 0,
|
|
|
|
|
+ Timestamp.from(record.getCreatedAt())
|
|
|
|
|
+ );
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// ─── SaveConversion ──────────────────────────────────────────────────────
|
|
// ─── SaveConversion ──────────────────────────────────────────────────────
|
|
|
|
|
|
|
|
public void saveConversion(ConversionRecord record) {
|
|
public void saveConversion(ConversionRecord record) {
|
|
|
if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
|
|
if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
|
|
|
- jdbc.update("""
|
|
|
|
|
- INSERT INTO baidu_conversions (
|
|
|
|
|
- dedupe_key, qk, media, date_value, appsid, customer_name, device_id, conv,
|
|
|
|
|
- payment, act, tu, clk_time, created_at
|
|
|
|
|
- ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
|
|
|
- ON DUPLICATE KEY UPDATE
|
|
|
|
|
- media = VALUES(media),
|
|
|
|
|
- conv = VALUES(conv),
|
|
|
|
|
- payment = VALUES(payment),
|
|
|
|
|
- clk_time = VALUES(clk_time)
|
|
|
|
|
- """,
|
|
|
|
|
- record.getDedupeKey(), record.getQk(), record.getMedia(),
|
|
|
|
|
- record.getDate(), record.getAppSid(), record.getCustomerName(),
|
|
|
|
|
- record.getDeviceId(), record.getConv(), record.getPayment(),
|
|
|
|
|
- record.getAct(), record.getTu(), record.getClkTime(),
|
|
|
|
|
- Timestamp.from(record.getCreatedAt())
|
|
|
|
|
- );
|
|
|
|
|
|
|
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
|
|
|
|
|
+ ConversionMapper mapper = session.getMapper(ConversionMapper.class);
|
|
|
|
|
+ mapper.insertOrUpdate(
|
|
|
|
|
+ record.getDedupeKey(), record.getQk(), record.getMedia(),
|
|
|
|
|
+ record.getDate(), record.getAppSid(), record.getCustomerName(),
|
|
|
|
|
+ record.getDeviceId(), record.getConv(), record.getPayment(),
|
|
|
|
|
+ record.getAct(), record.getTu(), record.getClkTime(),
|
|
|
|
|
+ Timestamp.from(record.getCreatedAt())
|
|
|
|
|
+ );
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// ─── SaveMediaCallback ───────────────────────────────────────────────────
|
|
// ─── SaveMediaCallback ───────────────────────────────────────────────────
|
|
@@ -206,87 +188,46 @@ public class TiDBColdStore {
|
|
|
public void saveMediaCallback(MediaCallbackRecord record) {
|
|
public void saveMediaCallback(MediaCallbackRecord record) {
|
|
|
if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
|
|
if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
|
|
|
int attempt = record.getAttempt() <= 0 ? 1 : record.getAttempt();
|
|
int attempt = record.getAttempt() <= 0 ? 1 : record.getAttempt();
|
|
|
- jdbc.update("""
|
|
|
|
|
- INSERT INTO media_callbacks (
|
|
|
|
|
- dedupe_key, media, qk, callback_url, event_type, event_time_ms, purchase,
|
|
|
|
|
- status, ok, attempt, response_body, error_message, created_at
|
|
|
|
|
- ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
|
|
|
- """,
|
|
|
|
|
- record.getDedupeKey(), record.getMedia(), record.getQk(),
|
|
|
|
|
- record.getCallbackUrl(), record.getEventType(), record.getEventTimeMs(),
|
|
|
|
|
- record.getPurchase(), record.getStatus(), record.isOk() ? 1 : 0,
|
|
|
|
|
- attempt, record.getResponseBody(), record.getErrorMessage(),
|
|
|
|
|
- Timestamp.from(record.getCreatedAt())
|
|
|
|
|
- );
|
|
|
|
|
|
|
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
|
|
|
|
|
+ MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
|
|
|
|
|
+ mapper.insert(
|
|
|
|
|
+ record.getDedupeKey(), record.getMedia(), record.getQk(),
|
|
|
|
|
+ record.getCallbackUrl(), record.getEventType(), record.getEventTimeMs(),
|
|
|
|
|
+ record.getPurchase(), record.getStatus(), record.isOk() ? 1 : 0,
|
|
|
|
|
+ attempt, record.getResponseBody(), record.getErrorMessage(),
|
|
|
|
|
+ Timestamp.from(record.getCreatedAt())
|
|
|
|
|
+ );
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// ─── 查询:成功回调是否存在 ────────────────────────────────────────────────
|
|
// ─── 查询:成功回调是否存在 ────────────────────────────────────────────────
|
|
|
|
|
|
|
|
public boolean successfulCallbackExists(String dedupeKey) {
|
|
public boolean successfulCallbackExists(String dedupeKey) {
|
|
|
if (dedupeKey == null || dedupeKey.isBlank()) return false;
|
|
if (dedupeKey == null || dedupeKey.isBlank()) return false;
|
|
|
- Integer count = jdbc.queryForObject(
|
|
|
|
|
- "SELECT COUNT(1) FROM media_callbacks WHERE dedupe_key = ? AND ok = 1 LIMIT 1",
|
|
|
|
|
- Integer.class, dedupeKey);
|
|
|
|
|
- return count != null && count > 0;
|
|
|
|
|
|
|
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
|
|
|
|
|
+ MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
|
|
|
|
|
+ int count = mapper.countSuccessfulByDedupeKey(dedupeKey);
|
|
|
|
|
+ return count > 0;
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
public boolean successfulCallbackExistsForEvent(String dedupeKey, String legacyDedupeKey, int eventType) {
|
|
public boolean successfulCallbackExistsForEvent(String dedupeKey, String legacyDedupeKey, int eventType) {
|
|
|
if (dedupeKey == null || dedupeKey.isBlank()) return false;
|
|
if (dedupeKey == null || dedupeKey.isBlank()) return false;
|
|
|
- Integer count = jdbc.queryForObject("""
|
|
|
|
|
- SELECT COUNT(1)
|
|
|
|
|
- FROM media_callbacks
|
|
|
|
|
- WHERE ok = 1
|
|
|
|
|
- AND (
|
|
|
|
|
- dedupe_key = ?
|
|
|
|
|
- OR (dedupe_key = ? AND event_type = ?)
|
|
|
|
|
- )
|
|
|
|
|
- LIMIT 1
|
|
|
|
|
- """, Integer.class, dedupeKey, legacyDedupeKey, eventType);
|
|
|
|
|
- return count != null && count > 0;
|
|
|
|
|
|
|
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
|
|
|
|
|
+ MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
|
|
|
|
|
+ int count = mapper.countSuccessfulForEvent(dedupeKey, legacyDedupeKey, eventType);
|
|
|
|
|
+ return count > 0;
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// ─── 查询:待重试的回调 ────────────────────────────────────────────────────
|
|
// ─── 查询:待重试的回调 ────────────────────────────────────────────────────
|
|
|
|
|
|
|
|
public List<MediaCallbackRecord> pendingMediaCallbacks(String media, int limit) {
|
|
public List<MediaCallbackRecord> pendingMediaCallbacks(String media, int limit) {
|
|
|
if (limit <= 0) limit = 100;
|
|
if (limit <= 0) limit = 100;
|
|
|
- return jdbc.query("""
|
|
|
|
|
- SELECT m.dedupe_key, m.media, m.qk, m.callback_url, m.event_type, m.event_time_ms,
|
|
|
|
|
- m.purchase, m.status, m.ok, m.attempt, m.response_body, m.error_message, m.created_at
|
|
|
|
|
- FROM media_callbacks m
|
|
|
|
|
- JOIN (
|
|
|
|
|
- SELECT dedupe_key, MAX(id) AS id
|
|
|
|
|
- FROM media_callbacks
|
|
|
|
|
- WHERE media = ? AND dedupe_key IS NOT NULL AND dedupe_key != ''
|
|
|
|
|
- GROUP BY dedupe_key
|
|
|
|
|
- ) latest ON latest.id = m.id
|
|
|
|
|
- WHERE m.media = ? AND m.ok = 0
|
|
|
|
|
- AND NOT EXISTS (
|
|
|
|
|
- SELECT 1 FROM media_callbacks ok_cb
|
|
|
|
|
- WHERE ok_cb.dedupe_key = m.dedupe_key AND ok_cb.ok = 1
|
|
|
|
|
- )
|
|
|
|
|
- ORDER BY m.created_at ASC
|
|
|
|
|
- LIMIT ?
|
|
|
|
|
- """,
|
|
|
|
|
- (rs, rowNum) -> {
|
|
|
|
|
- MediaCallbackRecord r = new MediaCallbackRecord();
|
|
|
|
|
- r.setDedupeKey(rs.getString("dedupe_key"));
|
|
|
|
|
- r.setMedia(rs.getString("media"));
|
|
|
|
|
- r.setQk(rs.getString("qk"));
|
|
|
|
|
- r.setCallbackUrl(rs.getString("callback_url"));
|
|
|
|
|
- r.setEventType(rs.getInt("event_type"));
|
|
|
|
|
- r.setEventTimeMs(rs.getLong("event_time_ms"));
|
|
|
|
|
- r.setPurchase(rs.getDouble("purchase"));
|
|
|
|
|
- r.setStatus(rs.getInt("status"));
|
|
|
|
|
- r.setOk(rs.getInt("ok") == 1);
|
|
|
|
|
- r.setAttempt(rs.getInt("attempt"));
|
|
|
|
|
- r.setResponseBody(rs.getString("response_body"));
|
|
|
|
|
- r.setErrorMessage(rs.getString("error_message"));
|
|
|
|
|
- Timestamp ts = rs.getTimestamp("created_at");
|
|
|
|
|
- if (ts != null) r.setCreatedAt(ts.toInstant());
|
|
|
|
|
- return r;
|
|
|
|
|
- },
|
|
|
|
|
- media, media, limit
|
|
|
|
|
- );
|
|
|
|
|
|
|
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
|
|
|
|
|
+ MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
|
|
|
|
|
+ return mapper.selectPendingCallbacks(media, limit);
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// ─── 工具方法 ─────────────────────────────────────────────────────────────
|
|
// ─── 工具方法 ─────────────────────────────────────────────────────────────
|