数据归档

ADX_REPORT_BACKFILL.md

@@ -0,0 +1,115 @@
+# adx_report 回灌说明
+
+本文档说明如何将 `tencent_ad_bid_events`、`honor_ad_bid_events`、`kuaishou_ad_bid_events` 的老历史数据归档到 `adx_report`，并生成日汇总表。
+
+## 1. 当前代码行为
+
+- 主库热表继续保留最近 30 天数据
+- 后台任务开启后，会把 30 天前数据按月搬到 `adx_report`
+- 月历史表命名规则：
+  - `tencent_ad_bid_events_history_YYYYMM`
+  - `honor_ad_bid_events_history_YYYYMM`
+  - `kuaishou_ad_bid_events_history_YYYYMM`
+- 日汇总表：
+  - `tencent_ad_bid_events_daily_stats`
+  - `honor_ad_bid_events_daily_stats`
+  - `kuaishou_ad_bid_events_daily_stats`
+
+## 2. 推荐上线顺序
+
+1. 先发版，让代码具备 `adx_report` 建库、归档、日汇总能力
+2. 在测试环境确认 `adx.ad-bid-report-enabled=true` 后运行正常
+3. 生产环境先保持 `adx.ad-bid-report-enabled=false`
+4. 确认主库 / 报表库实例容量、慢 SQL、锁冲突可接受后，再开启生产开关
+
+## 3. 一次性历史归档建议
+
+生产环境如果已经积累了大量超过 30 天的老数据，建议不要一开开关就直接放大批量归档。
+
+推荐分阶段：
+
+1. 先把生产配置中的批次控制调小
+
+```yaml
+adx:
+  ad-bid-report-enabled: true
+  ad-bid-report-archive-batch-size: 2000
+  ad-bid-report-archive-max-batches-per-run: 5
+```
+
+2. 观察 1 到 2 个归档周期，确认：
+   - 主库写入延迟正常
+   - `adx_report` 月表创建正常
+   - 没有异常锁等待
+
+3. 再逐步提高：
+
+```yaml
+adx:
+  ad-bid-report-archive-batch-size: 5000
+  ad-bid-report-archive-max-batches-per-run: 20
+```
+
+## 4. 日汇总补数说明
+
+当前代码会按 `adx.ad-bid-report-stats-lookback-days` 回算热表近 N 天数据并 upsert 到日汇总表。
+
+当前 daily stats 统计维度为：
+
+- `stat_date`
+- `platform`
+- `account_id`
+- `tag_id`
+- `event_type`
+
+当前 daily stats 指标为：
+
+- `bid_count`：该维度下的总事件数
+- `impression_count`：曝光数
+- `click_count`：点击数
+- `unique_media_trace_count`：按 `media_trace_id` 去重后的事件数
+- `price_sum`：`price > 0` 的记录数
+
+说明：
+
+- 汇总源数据来自三张热表，不直接扫月历史表
+- 统计窗口不包含当天数据，只统计到昨天 24:00 为止
+- 每次任务都会重算 lookback 窗口内的数据，再通过 upsert 覆盖旧值
+- 所以近 30 个自然日如果有补写、重复清理、延迟写入，daily stats 会被自动修正
+
+默认值：
+
+```yaml
+adx:
+  ad-bid-report-stats-lookback-days: 30
+```
+
+这意味着：
+
+- 最近 30 天统计会自动修正
+- 更久远的老历史不会自动全量重刷 daily stats
+
+如果需要补很久以前的 daily stats，建议单独执行离线 SQL 按月回灌。
+
+## 5. 建议核对项
+
+开启后建议核对：
+
+1. `adx_report` 是否自动创建
+2. 月历史表是否按 `YYYYMM` 创建
+3. 三张 daily stats 表是否有数据
+4. 主库热表 30 天前数据是否开始下降
+5. 在线链路是否仍只查热表，没有影响监测和回传
+
+## 6. 关键配置
+
+```yaml
+adx:
+  report-database-name: "adx_report"
+  ad-bid-report-enabled: false
+  ad-bid-report-interval: 30m
+  ad-bid-report-retention-days: 30
+  ad-bid-report-archive-batch-size: 5000
+  ad-bid-report-archive-max-batches-per-run: 20
+  ad-bid-report-stats-lookback-days: 30
+```

src/main/java/com/adx/tencent/AppConfiguration.java

@@ -27,6 +27,7 @@ import com.adx.tencent.kuaishou.service.KuaishouTagEventSyncService;
 import com.adx.tencent.kuaishou.store.KuaishouColdStore;
 import com.adx.tencent.kuaishou.store.KuaishouHotStore;
 import com.adx.tencent.leader.LeaderElection;
+import com.adx.tencent.report.AdBidReportStore;
 import com.adx.tencent.tagsync.TagEventResolver;
 import com.adx.tencent.tagsync.TagEventSyncService;
 import com.adx.tencent.tencent.TencentClient;
@@ -120,9 +121,11 @@ public class AppConfiguration {
     }
 
     @Bean
-    public RedisHotStore redisHotStore(StringRedisTemplate redis, ObjectMapper objectMapper) {
+    public RedisHotStore redisHotStore(StringRedisTemplate redis,
+                                       ObjectMapper objectMapper,
+                                       @Nullable TiDBColdStore coldStore) {
         return new RedisHotStore(redis, objectMapper,
-                tiDBColdStore(objectMapper),
+                coldStore,
                 "adx:tencent:",
                 props.getRedisStream(),
                 props.getBidTtl());
@@ -141,15 +144,28 @@ public class AppConfiguration {
     // ─── TiDB（可选）────────────────────────────────────────────────────────
 
     @Bean
-    public TiDBColdStore tiDBColdStore(ObjectMapper objectMapper) {
+    public DataSource tidbDataSource() {
         String url = props.getTidbUrl();
         if (url == null || url.isBlank()) {
             log.info("TiDB URL is empty; cold storage disabled");
             return null;
         }
         try {
-            DataSource ds = createDataSource(url, props.getTidbUsername(), props.getTidbPassword());
-            SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(ds);
+            return createDataSource("adx-tidb", url, props.getTidbUsername(), props.getTidbPassword());
+        } catch (Exception e) {
+            log.warn("Failed to open TiDB datasource: {}", e.getMessage(), e);
+            return null;
+        }
+    }
+
+    @Bean
+    public TiDBColdStore tiDBColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
+        if (tidbDataSource == null) {
+            log.info("TiDB datasource is empty; cold storage disabled");
+            return null;
+        }
+        try {
+            SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
             TiDBColdStore store = new TiDBColdStore(sqlSessionFactory, objectMapper);
             store.migrate();
             log.info("TiDB cold store initialized and migrated");
@@ -168,13 +184,13 @@ public class AppConfiguration {
         return factoryBean.getObject();
     }
 
-    private DataSource createDataSource(String jdbcUrl, String username, String password) {
+    private DataSource createDataSource(String poolName, String jdbcUrl, String username, String password) {
         HikariConfig config = new HikariConfig();
         config.setDriverClassName("com.mysql.cj.jdbc.Driver");
         config.setJdbcUrl(jdbcUrl);
         if (username != null && !username.isBlank()) config.setUsername(username);
         if (password != null && !password.isBlank()) config.setPassword(password);
-        config.setPoolName("adx-tidb");
+        config.setPoolName(poolName);
         config.setMinimumIdle(props.getTidbMinimumIdle());
         config.setMaximumPoolSize(props.getTidbMaximumPoolSize());
         config.setConnectionTimeout(props.getTidbConnectionTimeout().toMillis());
@@ -210,12 +226,10 @@ public class AppConfiguration {
     }
 
     @Bean
-    public HonorColdStore honorColdStore(ObjectMapper objectMapper) {
-        String url = props.getTidbUrl();
-        if (url == null || url.isBlank()) return null;
+    public HonorColdStore honorColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
+        if (tidbDataSource == null) return null;
         try {
-            DataSource ds = createDataSource(url, props.getTidbUsername(), props.getTidbPassword());
-            SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(ds);
+            SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
             HonorColdStore store = new HonorColdStore(sqlSessionFactory, objectMapper);
             store.migrate();
             return store;
@@ -286,12 +300,10 @@ public class AppConfiguration {
     }
 
     @Bean
-    public KuaishouColdStore kuaishouColdStore(ObjectMapper objectMapper) {
-        String url = props.getTidbUrl();
-        if (url == null || url.isBlank()) return null;
+    public KuaishouColdStore kuaishouColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
+        if (tidbDataSource == null) return null;
         try {
-            DataSource ds = createDataSource(url, props.getTidbUsername(), props.getTidbPassword());
-            SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(ds);
+            SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
             KuaishouColdStore store = new KuaishouColdStore(sqlSessionFactory, objectMapper);
             store.migrate();
             return store;
@@ -477,6 +489,23 @@ public class AppConfiguration {
                 props.getKuaishouTagEventSyncRedisPrefix(), ttl);
     }
 
+    @Bean
+    public AdBidReportStore adBidReportStore(@Nullable DataSource tidbDataSource) {
+        if (tidbDataSource == null) return null;
+        String mainSchema = extractDatabaseName(props.getTidbUrl());
+        if (mainSchema == null || mainSchema.isBlank()) return null;
+        AdBidReportStore store = new AdBidReportStore(
+                tidbDataSource,
+                mainSchema,
+                props.getReportDatabaseName(),
+                props.getAdBidReportRetentionDays(),
+                props.getAdBidReportArchiveBatchSize(),
+                props.getAdBidReportArchiveMaxBatchesPerRun(),
+                props.getAdBidReportStatsLookbackDays());
+        store.migrate();
+        return store;
+    }
+
     // ─── ColdWorker ─────────────────────────────────────────────────────────
 
     @Bean
@@ -517,6 +546,19 @@ public class AppConfiguration {
         return p;
     }
 
+    private static String extractDatabaseName(String jdbcUrl) {
+        if (jdbcUrl == null || jdbcUrl.isBlank()) {
+            return null;
+        }
+        int slash = jdbcUrl.lastIndexOf('/');
+        if (slash < 0 || slash + 1 >= jdbcUrl.length()) {
+            return null;
+        }
+        int question = jdbcUrl.indexOf('?', slash + 1);
+        String db = question >= 0 ? jdbcUrl.substring(slash + 1, question) : jdbcUrl.substring(slash + 1);
+        return db == null || db.isBlank() ? null : db.trim();
+    }
+
     // ─── 后台任务调度 ────────────────────────────────────────────────────────
 
     @Component

src/main/java/com/adx/tencent/config/AppProperties.java

@@ -99,6 +99,13 @@ public class AppProperties {
     private Duration tidbIdleTimeout = Duration.ofMinutes(5);
     private Duration tidbMaxLifetime = Duration.ofMinutes(10);
     private Duration tidbKeepaliveTime = Duration.ofMinutes(2);
+    private String reportDatabaseName = "adx_report";
+    private boolean adBidReportEnabled = false;
+    private Duration adBidReportInterval = Duration.ofMinutes(30);
+    private int adBidReportRetentionDays = 30;
+    private int adBidReportArchiveBatchSize = 5000;
+    private int adBidReportArchiveMaxBatchesPerRun = 20;
+    private int adBidReportStatsLookbackDays = 30;
 
     // Task Lock
     private Duration taskLockTtl = Duration.ofSeconds(120);
@@ -674,6 +681,62 @@ public class AppProperties {
         this.tidbKeepaliveTime = v;
     }
 
+    public String getReportDatabaseName() {
+        return reportDatabaseName;
+    }
+
+    public void setReportDatabaseName(String v) {
+        this.reportDatabaseName = v;
+    }
+
+    public boolean isAdBidReportEnabled() {
+        return adBidReportEnabled;
+    }
+
+    public void setAdBidReportEnabled(boolean v) {
+        this.adBidReportEnabled = v;
+    }
+
+    public Duration getAdBidReportInterval() {
+        return adBidReportInterval;
+    }
+
+    public void setAdBidReportInterval(Duration v) {
+        this.adBidReportInterval = v;
+    }
+
+    public int getAdBidReportRetentionDays() {
+        return adBidReportRetentionDays;
+    }
+
+    public void setAdBidReportRetentionDays(int v) {
+        this.adBidReportRetentionDays = v;
+    }
+
+    public int getAdBidReportArchiveBatchSize() {
+        return adBidReportArchiveBatchSize;
+    }
+
+    public void setAdBidReportArchiveBatchSize(int v) {
+        this.adBidReportArchiveBatchSize = v;
+    }
+
+    public int getAdBidReportArchiveMaxBatchesPerRun() {
+        return adBidReportArchiveMaxBatchesPerRun;
+    }
+
+    public void setAdBidReportArchiveMaxBatchesPerRun(int v) {
+        this.adBidReportArchiveMaxBatchesPerRun = v;
+    }
+
+    public int getAdBidReportStatsLookbackDays() {
+        return adBidReportStatsLookbackDays;
+    }
+
+    public void setAdBidReportStatsLookbackDays(int v) {
+        this.adBidReportStatsLookbackDays = v;
+    }
+
     public Duration getTaskLockTtl() {
         return taskLockTtl;
     }

src/main/java/com/adx/tencent/report/AdBidReportBackgroundTasks.java

@@ -0,0 +1,81 @@
+package com.adx.tencent.report;
+
+import com.adx.tencent.config.AppProperties;
+import com.adx.tencent.leader.LeaderElection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.lang.Nullable;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@Component
+public class AdBidReportBackgroundTasks implements ApplicationRunner {
+
+    private static final Logger log = LoggerFactory.getLogger(AdBidReportBackgroundTasks.class);
+
+    @Autowired private AppProperties props;
+    @Autowired(required = false) private AdBidReportStore adBidReportStore;
+    @Autowired(required = false) @Nullable private LeaderElection leaderElection;
+
+    private final AtomicBoolean stopped = new AtomicBoolean(false);
+    private final ExecutorService executor = Executors.newCachedThreadPool(r -> {
+        Thread t = new Thread(r);
+        t.setDaemon(true);
+        return t;
+    });
+
+    @Override
+    public void run(ApplicationArguments args) {
+        if (adBidReportStore == null || !props.isAdBidReportEnabled()) {
+            return;
+        }
+        if (props.isSkipLeaderElection() || leaderElection == null) {
+            executor.submit(() -> runReportJob(() -> stopped.get()));
+        } else {
+            executor.submit(() -> leaderElection.run(
+                    "adx:lock:report:ad-bid-archive",
+                    props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
+                    stopped::get,
+                    this::runReportJob,
+                    e -> log.error("ad bid report leader: {}", e.getMessage(), e)
+            ));
+        }
+    }
+
+    private void runReportJob(LeaderElection.StopSignal stopSignal) {
+        long intervalMs = props.getAdBidReportInterval().toMillis();
+        while (!stopSignal.isStopped()) {
+            try {
+                AdBidReportStore.RunSummary summary = adBidReportStore.runOnce();
+                log.info("[AdBidReport] done | archivedRows={} | statsRows={}",
+                        summary.archivedRows, summary.statsRows);
+            } catch (Exception e) {
+                log.error("[AdBidReport] error: {}", e.getMessage(), e);
+            }
+            sleepResponsive(intervalMs, stopSignal);
+        }
+    }
+
+    private void sleepResponsive(long millis, LeaderElection.StopSignal stopSignal) {
+        long remaining = millis;
+        while (remaining > 0 && !stopSignal.isStopped()) {
+            long chunk = Math.min(remaining, 1000);
+            sleep(chunk);
+            remaining -= chunk;
+        }
+    }
+
+    private static void sleep(long millis) {
+        try {
+            Thread.sleep(millis);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}

src/main/java/com/adx/tencent/report/AdBidReportStore.java

@@ -0,0 +1,270 @@
+package com.adx.tencent.report;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.YearMonth;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+
+public class AdBidReportStore {
+
+    private static final ZoneId ZONE = ZoneId.of("Asia/Shanghai");
+
+    private final DataSource dataSource;
+    private final String mainSchema;
+    private final String reportSchema;
+    private final int retentionDays;
+    private final int archiveBatchSize;
+    private final int archiveMaxBatchesPerRun;
+    private final int statsLookbackDays;
+
+    public AdBidReportStore(DataSource dataSource,
+                            String mainSchema,
+                            String reportSchema,
+                            int retentionDays,
+                            int archiveBatchSize,
+                            int archiveMaxBatchesPerRun,
+                            int statsLookbackDays) {
+        this.dataSource = dataSource;
+        this.mainSchema = safeIdentifier(mainSchema);
+        this.reportSchema = safeIdentifier(reportSchema == null || reportSchema.isBlank() ? "adx_report" : reportSchema);
+        this.retentionDays = Math.max(retentionDays, 1);
+        this.archiveBatchSize = Math.max(archiveBatchSize, 100);
+        this.archiveMaxBatchesPerRun = Math.max(archiveMaxBatchesPerRun, 1);
+        this.statsLookbackDays = Math.max(statsLookbackDays, 1);
+    }
+
+    public void migrate() {
+        try (Connection conn = dataSource.getConnection();
+             PreparedStatement createDb = conn.prepareStatement(
+                     "CREATE DATABASE IF NOT EXISTS `" + reportSchema + "` DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci")) {
+            createDb.execute();
+            for (MediaTable media : MediaTable.values()) {
+                try (PreparedStatement ps = conn.prepareStatement(createDailyStatsDdl(media))) {
+                    ps.execute();
+                }
+                ensureDailyStatsColumns(conn, media);
+            }
+        } catch (Exception e) {
+            throw new RuntimeException("ad bid report migrate failed", e);
+        }
+    }
+
+    public RunSummary runOnce() {
+        RunSummary summary = new RunSummary();
+        Timestamp cutoff = Timestamp.from(Instant.now().minusSeconds(retentionDays * 24L * 3600L));
+        try (Connection conn = dataSource.getConnection()) {
+            conn.setAutoCommit(false);
+            for (MediaTable media : MediaTable.values()) {
+                summary.archivedRows += archiveMedia(conn, media, cutoff);
+                summary.statsRows += rollupStats(conn, media);
+            }
+            conn.commit();
+            return summary;
+        } catch (Exception e) {
+            throw new RuntimeException("ad bid report run failed", e);
+        }
+    }
+
+    private long archiveMedia(Connection conn, MediaTable media, Timestamp cutoff) throws Exception {
+        List<YearMonth> months = listArchivableMonths(conn, media, cutoff);
+        long archived = 0L;
+        for (YearMonth month : months) {
+            ensureHistoryTable(conn, media, month);
+            Timestamp monthStart = Timestamp.valueOf(month.atDay(1).atStartOfDay());
+            Timestamp monthEnd = Timestamp.valueOf(month.plusMonths(1).atDay(1).atStartOfDay());
+            for (int i = 0; i < archiveMaxBatchesPerRun; i++) {
+                int inserted = insertHistoryBatch(conn, media, month, monthStart, monthEnd, cutoff);
+                int deleted = deleteHotBatch(conn, media, monthStart, monthEnd, cutoff);
+                if (deleted > 0) {
+                    archived += deleted;
+                }
+                if (inserted == 0 && deleted == 0) {
+                    break;
+                }
+            }
+        }
+        return archived;
+    }
+
+    private List<YearMonth> listArchivableMonths(Connection conn, MediaTable media, Timestamp cutoff) throws Exception {
+        String sql = "SELECT DATE_FORMAT(created_at, '%Y%m') AS ym FROM `" + mainSchema + "`.`" + media.hotTable + "` " +
+                "WHERE created_at < ? GROUP BY ym ORDER BY ym ASC";
+        List<YearMonth> result = new ArrayList<>();
+        try (PreparedStatement ps = conn.prepareStatement(sql)) {
+            ps.setTimestamp(1, cutoff);
+            try (ResultSet rs = ps.executeQuery()) {
+                while (rs.next()) {
+                    String ym = rs.getString("ym");
+                    if (ym != null && ym.length() == 6) {
+                        result.add(YearMonth.of(Integer.parseInt(ym.substring(0, 4)), Integer.parseInt(ym.substring(4, 6))));
+                    }
+                }
+            }
+        }
+        return result;
+    }
+
+    private void ensureHistoryTable(Connection conn, MediaTable media, YearMonth month) throws Exception {
+        String table = historyTable(media, month);
+        String sql = "CREATE TABLE IF NOT EXISTS `" + reportSchema + "`.`" + table + "` LIKE `" +
+                mainSchema + "`.`" + media.hotTable + "`";
+        try (PreparedStatement ps = conn.prepareStatement(sql)) {
+            ps.execute();
+        }
+    }
+
+    private int insertHistoryBatch(Connection conn,
+                                   MediaTable media,
+                                   YearMonth month,
+                                   Timestamp monthStart,
+                                   Timestamp monthEnd,
+                                   Timestamp cutoff) throws Exception {
+        String table = historyTable(media, month);
+        String sql = "INSERT IGNORE INTO `" + reportSchema + "`.`" + table + "` " +
+                "SELECT * FROM `" + mainSchema + "`.`" + media.hotTable + "` " +
+                "WHERE created_at >= ? AND created_at < ? AND created_at < ? " +
+                "ORDER BY created_at ASC LIMIT ?";
+        try (PreparedStatement ps = conn.prepareStatement(sql)) {
+            ps.setTimestamp(1, monthStart);
+            ps.setTimestamp(2, monthEnd);
+            ps.setTimestamp(3, cutoff);
+            ps.setInt(4, archiveBatchSize);
+            return ps.executeUpdate();
+        }
+    }
+
+    private int deleteHotBatch(Connection conn,
+                               MediaTable media,
+                               Timestamp monthStart,
+                               Timestamp monthEnd,
+                               Timestamp cutoff) throws Exception {
+        String sql = "DELETE FROM `" + mainSchema + "`.`" + media.hotTable + "` " +
+                "WHERE created_at >= ? AND created_at < ? AND created_at < ? " +
+                "ORDER BY created_at ASC LIMIT ?";
+        try (PreparedStatement ps = conn.prepareStatement(sql)) {
+            ps.setTimestamp(1, monthStart);
+            ps.setTimestamp(2, monthEnd);
+            ps.setTimestamp(3, cutoff);
+            ps.setInt(4, archiveBatchSize);
+            return ps.executeUpdate();
+        }
+    }
+
+    private long rollupStats(Connection conn, MediaTable media) throws Exception {
+        LocalDate endExclusiveDate = LocalDate.now(ZONE);
+        LocalDate startDate = endExclusiveDate.minusDays(statsLookbackDays);
+        Timestamp start = Timestamp.valueOf(startDate.atStartOfDay());
+        Timestamp end = Timestamp.valueOf(endExclusiveDate.atStartOfDay());
+        String sql = "INSERT INTO `" + reportSchema + "`.`" + media.statsTable + "` " +
+                "(stat_date, platform, account_id, tag_id, event_type, bid_count, impression_count, click_count, unique_media_trace_count, price_sum, created_at, updated_at) " +
+                "SELECT DATE(created_at) AS stat_date, " +
+                "COALESCE(platform, '') AS platform, " +
+                "COALESCE(account_id, '') AS account_id, " +
+                "COALESCE(tag_id, '') AS tag_id, " +
+                "COALESCE(event_type, '') AS event_type, " +
+                "COUNT(*) AS bid_count, " +
+                "SUM(CASE WHEN COALESCE(event_type, '') = 'impression' THEN 1 ELSE 0 END) AS impression_count, " +
+                "SUM(CASE WHEN COALESCE(event_type, '') = 'click' THEN 1 ELSE 0 END) AS click_count, " +
+                "COUNT(DISTINCT CASE WHEN media_trace_id IS NULL OR media_trace_id = '' THEN NULL ELSE media_trace_id END) AS unique_media_trace_count, " +
+                "SUM(CASE WHEN COALESCE(price, 0) > 0 THEN 1 ELSE 0 END) AS price_sum, " +
+                "CURRENT_TIMESTAMP(3), CURRENT_TIMESTAMP(3) " +
+                "FROM `" + mainSchema + "`.`" + media.hotTable + "` " +
+                "WHERE created_at >= ? AND created_at < ? " +
+                "GROUP BY DATE(created_at), COALESCE(platform, ''), COALESCE(account_id, ''), COALESCE(tag_id, ''), COALESCE(event_type, '') " +
+                "ON DUPLICATE KEY UPDATE " +
+                "bid_count = VALUES(bid_count), " +
+                "impression_count = VALUES(impression_count), " +
+                "click_count = VALUES(click_count), " +
+                "unique_media_trace_count = VALUES(unique_media_trace_count), " +
+                "price_sum = VALUES(price_sum), " +
+                "updated_at = VALUES(updated_at)";
+        try (PreparedStatement ps = conn.prepareStatement(sql)) {
+            ps.setTimestamp(1, start);
+            ps.setTimestamp(2, end);
+            return ps.executeUpdate();
+        }
+    }
+
+    private String historyTable(MediaTable media, YearMonth month) {
+        return media.historyPrefix + month.format(java.time.format.DateTimeFormatter.ofPattern("yyyyMM", Locale.ROOT));
+    }
+
+    private String createDailyStatsDdl(MediaTable media) {
+        return "CREATE TABLE IF NOT EXISTS `" + reportSchema + "`.`" + media.statsTable + "` (" +
+                "id BIGINT AUTO_INCREMENT PRIMARY KEY," +
+                "stat_date DATE NOT NULL COMMENT '统计日期'," +
+                "platform VARCHAR(16) NOT NULL DEFAULT '' COMMENT '平台(android/ios)'," +
+                "account_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '账户ID'," +
+                "tag_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '百度广告位ID'," +
+                "event_type VARCHAR(32) NOT NULL DEFAULT '' COMMENT '事件类型(impression/click)'," +
+                "bid_count BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '竞价事件数'," +
+                "impression_count BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '曝光事件数'," +
+                "click_count BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '点击事件数'," +
+                "unique_media_trace_count BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '去重媒体追踪ID数'," +
+                "price_sum BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT 'price>0记录数'," +
+                "created_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间'," +
+                "updated_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '更新时间'," +
+                "UNIQUE KEY `uk_" + media.statsTable + "_dims` (stat_date, platform, account_id, tag_id, event_type)," +
+                "KEY `idx_" + media.statsTable + "_tag_date` (tag_id, stat_date)," +
+                "KEY `idx_" + media.statsTable + "_account_date` (account_id, stat_date)" +
+                ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='" + media.comment + "日汇总统计'";
+    }
+
+    private void ensureDailyStatsColumns(Connection conn, MediaTable media) {
+        tryAlter(conn, "ALTER TABLE `" + reportSchema + "`.`" + media.statsTable + "` " +
+                "ADD COLUMN impression_count BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '曝光事件数' AFTER bid_count");
+        tryAlter(conn, "ALTER TABLE `" + reportSchema + "`.`" + media.statsTable + "` " +
+                "ADD COLUMN click_count BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '点击事件数' AFTER impression_count");
+        tryAlter(conn, "ALTER TABLE `" + reportSchema + "`.`" + media.statsTable + "` " +
+                "ADD COLUMN unique_media_trace_count BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '去重媒体追踪ID数' AFTER click_count");
+        tryAlter(conn, "ALTER TABLE `" + reportSchema + "`.`" + media.statsTable + "` " +
+                "ADD COLUMN price_sum BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT 'price>0记录数' AFTER unique_media_trace_count");
+        tryAlter(conn, "ALTER TABLE `" + reportSchema + "`.`" + media.statsTable + "` " +
+                "MODIFY COLUMN price_sum BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT 'price>0记录数'");
+    }
+
+    private static void tryAlter(Connection conn, String sql) {
+        try (PreparedStatement ps = conn.prepareStatement(sql)) {
+            ps.execute();
+        } catch (Exception ignored) {
+        }
+    }
+
+    private static String safeIdentifier(String value) {
+        if (value == null || !value.matches("[A-Za-z0-9_]+")) {
+            throw new IllegalArgumentException("invalid sql identifier: " + value);
+        }
+        return value;
+    }
+
+    public static class RunSummary {
+        public long archivedRows;
+        public long statsRows;
+    }
+
+    private enum MediaTable {
+        TENCENT("tencent_ad_bid_events", "tencent_ad_bid_events_history_", "tencent_ad_bid_events_daily_stats", "腾讯"),
+        HONOR("honor_ad_bid_events", "honor_ad_bid_events_history_", "honor_ad_bid_events_daily_stats", "荣耀"),
+        KUAISHOU("kuaishou_ad_bid_events", "kuaishou_ad_bid_events_history_", "kuaishou_ad_bid_events_daily_stats", "快手");
+
+        private final String hotTable;
+        private final String historyPrefix;
+        private final String statsTable;
+        private final String comment;
+
+        MediaTable(String hotTable, String historyPrefix, String statsTable, String comment) {
+            this.hotTable = hotTable;
+            this.historyPrefix = historyPrefix;
+            this.statsTable = statsTable;
+            this.comment = comment;
+        }
+    }
+}

src/main/resources/application-dev.yml

@@ -61,6 +61,13 @@ adx:
   tidb-idle-timeout: 5m
   tidb-max-lifetime: 10m
   tidb-keepalive-time: 2m
+  report-database-name: "adx_report"
+  ad-bid-report-enabled: false
+  ad-bid-report-interval: 30m
+  ad-bid-report-retention-days: 30
+  ad-bid-report-archive-batch-size: 5000
+  ad-bid-report-archive-max-batches-per-run: 20
+  ad-bid-report-stats-lookback-days: 30
 
   # --- Task ---
   skip-leader-election: false

src/main/resources/application-prod.yml

@@ -63,6 +63,13 @@ adx:
   tidb-idle-timeout: 90s
   tidb-max-lifetime: 2m
   tidb-keepalive-time: 30s
+  report-database-name: "adx_report"
+  ad-bid-report-enabled: true
+  ad-bid-report-interval: 30m
+  ad-bid-report-retention-days: 30
+  ad-bid-report-archive-batch-size: 5000
+  ad-bid-report-archive-max-batches-per-run: 20
+  ad-bid-report-stats-lookback-days: 30
 
   # --- Task ---
   skip-leader-election: false

src/main/resources/application-test.yml

@@ -62,6 +62,13 @@ adx:
   tidb-idle-timeout: 5m
   tidb-max-lifetime: 10m
   tidb-keepalive-time: 2m
+  report-database-name: "adx_report"
+  ad-bid-report-enabled: true
+  ad-bid-report-interval: 30m
+  ad-bid-report-retention-days: 30
+  ad-bid-report-archive-batch-size: 5000
+  ad-bid-report-archive-max-batches-per-run: 20
+  ad-bid-report-stats-lookback-days: 30
 
   # --- Task ---
   skip-leader-election: false