tencent-baidu-tracking/src/main/java/com/adx/tencent/AppConfiguration.java

package com.adx.tencent;

import com.adx.tencent.baidu.AdxClient;
import com.adx.tencent.baidu.AuctionPriceEncoder;
import com.adx.tencent.baidu.ConversionClient;
import com.adx.tencent.config.AppProperties;
import com.adx.tencent.conversionsync.ConversionSyncRunner;
import com.adx.tencent.conversionsync.ConversionSyncService;
import com.adx.tencent.conversionsync.ConversionBackfillJobService;
import com.adx.tencent.conversionsync.RetryService;
import com.adx.tencent.httpapi.MediaPlacement;
import com.adx.tencent.leader.LeaderElection;
import com.adx.tencent.tagsync.TagEventResolver;
import com.adx.tencent.tagsync.TagEventSyncService;
import com.adx.tencent.tencent.TencentClient;
import com.adx.tencent.storage.RedisHotStore;
import com.adx.tencent.storage.RedisLock;
import com.adx.tencent.storage.TiDBColdStore;
import com.adx.tencent.worker.ColdWorker;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.scheduling.annotation.Scheduled;

import org.springframework.lang.Nullable;

import javax.sql.DataSource;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 腾讯-百度回传项目 Bean 组装配置类。
 * 负责所有核心 Bean 的创建和后台任务（Worker、Leader 选举、定时同步等）。
 */
@Configuration
public class AppConfiguration {

    private static final Logger log = LoggerFactory.getLogger(AppConfiguration.class);

    @Autowired private AppProperties props;

    // ─── Redis ───────────────────────────────────────────────────────────────

    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        String addr = props.getRedisAddr();
        String[] parts = addr.split(":", 2);
        String host = parts[0];
        int port = parts.length > 1 ? Integer.parseInt(parts[1]) : 6379;

        RedisStandaloneConfiguration cfg = new RedisStandaloneConfiguration(host, port);
        cfg.setDatabase(props.getRedisDb());
        if (props.getRedisUsername() != null && !props.getRedisUsername().isBlank()) {
            cfg.setUsername(props.getRedisUsername());
        }
        if (props.getRedisPassword() != null && !props.getRedisPassword().isBlank()) {
            cfg.setPassword(props.getRedisPassword());
        }
        return new LettuceConnectionFactory(cfg);
    }

    @Bean
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {
        return new StringRedisTemplate(factory);
    }

    @Bean
    public RedisHotStore redisHotStore(StringRedisTemplate redis, ObjectMapper objectMapper) {
        return new RedisHotStore(redis, objectMapper,
                tiDBColdStore(objectMapper),
                "adx:tencent:",
                props.getRedisStream(),
                props.getBidTtl());
    }

    @Bean
    public RedisLock redisLock(StringRedisTemplate redis) {
        return new RedisLock(redis);
    }

    @Bean
    public LeaderElection leaderElection(RedisLock redisLock) {
        return new LeaderElection(redisLock);
    }

    // ─── TiDB（可选）────────────────────────────────────────────────────────

    @Bean
    public TiDBColdStore tiDBColdStore(ObjectMapper objectMapper) {
        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);
            TiDBColdStore store = new TiDBColdStore(sqlSessionFactory, objectMapper);
            store.migrate();
            log.info("TiDB cold store initialized and migrated");
            return store;
        } catch (Exception e) {
            log.warn("Failed to open TiDB: {}; cold storage disabled", e.getMessage(), e);
            return null;
        }
    }

    private SqlSessionFactory createSqlSessionFactory(DataSource dataSource) throws Exception {
        SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
        factoryBean.setDataSource(dataSource);
        factoryBean.setMapperLocations(
                new PathMatchingResourcePatternResolver().getResources("classpath:mapper/*.xml"));
        return factoryBean.getObject();
    }

    private DataSource createDataSource(String jdbcUrl, String username, String password) {
        org.springframework.jdbc.datasource.DriverManagerDataSource ds =
                new org.springframework.jdbc.datasource.DriverManagerDataSource();
        ds.setDriverClassName("com.mysql.cj.jdbc.Driver");
        ds.setUrl(jdbcUrl);
        if (username != null && !username.isBlank()) ds.setUsername(username);
        if (password != null && !password.isBlank()) ds.setPassword(password);
        return ds;
    }

    // ─── Baidu 客户端 ────────────────────────────────────────────────────────

    @Bean
    public AuctionPriceEncoder auctionPriceEncoder() {
        return AuctionPriceEncoder.create(
                props.getBaiduAuctionPriceEncryption(),
                props.getBaiduAuctionPriceEKey(),
                props.getBaiduAuctionPriceIKey());
    }

    @Bean
    public AdxClient adxClient(AuctionPriceEncoder encoder, ObjectMapper objectMapper) {
        return new AdxClient(props.getBaiduAdxEndpoint(), encoder, objectMapper);
    }

    @Bean
    public ConversionClient conversionClient(ObjectMapper objectMapper) {
        return new ConversionClient(
                props.getBaiduConversionBaseUrl(),
                props.getBaiduCustomerName(),
                props.getBaiduSecret(),
                objectMapper);
    }

    // ─── Tencent 客户端 + Sync 服务 ──────────────────────────────────────────

    @Bean
    public TencentClient tencentClient(ObjectMapper objectMapper, StringRedisTemplate redisTemplate) {
        // token 从 Redis 动态获取，支持外部刷新
        String tokenRedisKey = props.getTencentAccessTokenRedisKey();
        java.util.function.Supplier<String> tokenSupplier;
        if (tokenRedisKey != null && !tokenRedisKey.isBlank()) {
            tokenSupplier = () -> redisTemplate.opsForValue().get(tokenRedisKey);
        } else {
            // 兜底：使用配置文件中的静态 token
            String staticToken = props.getTencentAccessToken();
            tokenSupplier = () -> staticToken;
        }
        return new TencentClient(
                tokenSupplier,
                props.getTencentActMap(),
                objectMapper);
    }

    @Bean
    public TagEventResolver tagEventResolver(@Nullable TiDBColdStore coldStore,
                                             StringRedisTemplate redisTemplate) {
        if (coldStore == null) return null;
        return new TagEventResolver(redisTemplate, coldStore, props.getTagEventSyncRedisPrefix());
    }

    @Bean
    public ConversionSyncService conversionSyncService(ConversionClient baiduClient,
                                                         RedisHotStore hotStore,
                                                         TencentClient tencentClient,
                                                         @Nullable TiDBColdStore coldStore,
                                                         @Nullable TagEventResolver tagEventResolver) {
        if (coldStore == null) return null;
        return new ConversionSyncService(baiduClient, hotStore, tencentClient, coldStore, tagEventResolver);
    }

    @Bean
    public ConversionSyncRunner conversionSyncRunner(@Nullable ConversionSyncService syncer) {
        if (syncer == null) return null;
        return new ConversionSyncRunner(syncer,
                props.getConversionSyncDateOffsetDays(),
                props.getConversionSyncPageSize(),
                props.getConversionSyncActs());
    }

    @Bean
    public ConversionBackfillJobService conversionBackfillJobService(@Nullable ConversionSyncRunner syncer) {
        if (syncer == null) return null;
        return new ConversionBackfillJobService(syncer);
    }

    @Bean
    public RetryService retryService(@Nullable TiDBColdStore coldStore, TencentClient tencentClient) {
        if (coldStore == null) return null;
        return new RetryService(coldStore, tencentClient, props.getCallbackRetryLimit());
    }

    @Bean
    public TagEventSyncService tagEventSyncService(@Nullable TiDBColdStore coldStore,
                                                   StringRedisTemplate redisTemplate) {
        if (coldStore == null) return null;
        // TTL 取同步间隔的 3 倍，避免表中已删除的 tag 在 Redis 长期残留
        Duration ttl = props.getTagEventSyncInterval().multipliedBy(3);
        return new TagEventSyncService(coldStore, redisTemplate,
                props.getTagEventSyncRedisPrefix(), ttl);
    }

    // ─── ColdWorker ─────────────────────────────────────────────────────────

    @Bean
    public ColdWorker coldWorker(RedisHotStore hotStore, @Nullable TiDBColdStore coldStore,
                                  ObjectMapper objectMapper) {
        if (coldStore == null) return null;
        return new ColdWorker(hotStore, coldStore,
                props.getWorkerGroup(), props.getWorkerConsumer(),
                props.getWorkerBatch(), null,
                props.getRedisStreamMaxLen(), objectMapper);
    }

    // ─── MediaPlacement（腾讯）───────────────────────────────────────────────

    @Bean
    public MediaPlacement tencentPlacement() {
        MediaPlacement p = new MediaPlacement();
        p.setBaiduMediaId(props.getBaiduMediaId());
        p.setBaiduAppId(props.getBaiduAppId());
        p.setBaiduTagId(props.getBaiduTagId());
        p.setBidFloor(props.getBaiduBidFloor());
        p.setActionTypes(List.of(0, 1, 2));

        Map<String, MediaPlacement.BaiduAppPlacement> platforms = new HashMap<>();
        if (props.getBaiduAndroidAppId() != null && !props.getBaiduAndroidAppId().isBlank()) {
            MediaPlacement.BaiduAppPlacement android = new MediaPlacement.BaiduAppPlacement();
            android.setAppId(props.getBaiduAndroidAppId());
            android.setTagIds(props.getBaiduAndroidTagIds());
            platforms.put("android", android);
        }
        if (props.getBaiduIosAppId() != null && !props.getBaiduIosAppId().isBlank()) {
            MediaPlacement.BaiduAppPlacement ios = new MediaPlacement.BaiduAppPlacement();
            ios.setAppId(props.getBaiduIosAppId());
            ios.setTagIds(props.getBaiduIosTagIds());
            platforms.put("ios", ios);
        }
        if (!platforms.isEmpty()) p.setPlatforms(platforms);
        return p;
    }

    // ─── 后台任务调度 ────────────────────────────────────────────────────────

    @Component
    static class BackgroundTasks implements ApplicationRunner {

        private static final Logger taskLog = LoggerFactory.getLogger(BackgroundTasks.class);

        @Autowired private AppProperties props;
        @Autowired(required = false) private ColdWorker coldWorker;
        @Autowired(required = false) private ConversionSyncRunner conversionSyncRunner;
        @Autowired(required = false) private RetryService retryService;
        @Autowired(required = false) private TagEventSyncService tagEventSyncService;
        @Autowired(required = false) 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) {
            // 冷路径 Worker
            if (coldWorker != null) {
                executor.submit(() -> {
                    while (!stopped.get()) {
                        try {
                            coldWorker.runOnce();
                        } catch (Exception e) {
                            taskLog.error("cold worker: {}", e.getMessage(), e);
                            sleep(1000);
                        }
                    }
                });
                taskLog.info("Cold worker started");
            }

            // 转化同步
            if (conversionSyncRunner != null && props.isConversionSyncEnabled()) {
                if (props.isSkipLeaderElection()) {
                    executor.submit(() -> runConversionSync(() -> stopped.get()));
                    taskLog.info("Conversion sync started (skip leader election)");
                } else if (leaderElection != null) {
                    executor.submit(() ->
                        leaderElection.run(
                            "adx:lock:tencent:conversion-sync",
                            props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
                            stopped::get,
                            (jobStop) -> runConversionSync(jobStop),
                            e -> taskLog.error("tencent conversion sync leader: {}", e.getMessage(), e)
                        )
                    );
                    taskLog.info("Conversion sync leader election started");
                }
            } else {
                taskLog.warn("Conversion sync NOT started: runner={}, enabled={}",
                        conversionSyncRunner != null, props.isConversionSyncEnabled());
            }

            // 回调重试
            if (retryService != null && props.isCallbackRetryEnabled()) {
                if (props.isSkipLeaderElection()) {
                    executor.submit(() -> runCallbackRetry(() -> stopped.get()));
                    taskLog.info("Callback retry started (skip leader election)");
                } else if (leaderElection != null) {
                    executor.submit(() ->
                        leaderElection.run(
                            "adx:lock:tencent:callback-retry",
                            props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
                            stopped::get,
                            (jobStop) -> runCallbackRetry(jobStop),
                            e -> taskLog.error("tencent callback retry leader: {}", e.getMessage(), e)
                        )
                    );
                    taskLog.info("Callback retry leader election started");
                }
            } else {
                taskLog.warn("Callback retry NOT started: retryService={}, enabled={}",
                        retryService != null, props.isCallbackRetryEnabled());
            }

            // 广告位回传方式同步
            if (tagEventSyncService != null && props.isTagEventSyncEnabled()) {
                if (props.isSkipLeaderElection()) {
                    executor.submit(() -> runTagEventSync(() -> stopped.get()));
                    taskLog.info("Tag event sync started (skip leader election)");
                } else if (leaderElection != null) {
                    executor.submit(() ->
                        leaderElection.run(
                            "adx:lock:tencent:tag-event-sync",
                            props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
                            stopped::get,
                            (jobStop) -> runTagEventSync(jobStop),
                            e -> taskLog.error("tencent tag event sync leader: {}", e.getMessage(), e)
                        )
                    );
                    taskLog.info("Tag event sync leader election started");
                }
            } else {
                taskLog.warn("Tag event sync NOT started: service={}, enabled={}",
                        tagEventSyncService != null, props.isTagEventSyncEnabled());
            }
        }

        private void runConversionSync(LeaderElection.StopSignal jobStop) {
            long intervalMs = props.getConversionSyncInterval().toMillis();
            taskLog.info("[ConversionSync] task started, interval={}ms", intervalMs);
            while (!jobStop.isStopped()) {
                try {
                    taskLog.info("[ConversionSync] executing...");
                    ConversionSyncService.SyncResult r = conversionSyncRunner.runOnce();
                    taskLog.info("[ConversionSync] done: fetched={} sent={} failed={}", r.fetched.get(), r.sent.get(), r.failed.get());
                } catch (Exception e) {
                    taskLog.error("[ConversionSync] error: {}", e.getMessage(), e);
                }
                taskLog.info("[ConversionSync] sleeping {}ms until next run", intervalMs);
                sleepResponsive(intervalMs, jobStop);
            }
            taskLog.info("[ConversionSync] task stopped");
        }

        @Scheduled(cron = "0 17 1,7,10,16,20,23 * * ?")
        public void runDelayedConversionSync() {
            if (conversionSyncRunner == null || !props.isConversionSyncEnabled()) {
                return;
            }
            if (stopped.get()) {
                return;
            }

            Runnable job = () -> {
                long startNs = System.nanoTime();
                try {
                    taskLog.info("[ConversionBackfill] executing for offsets=1..7");
                    ConversionSyncService.SyncResult r = conversionSyncRunner.runBackfillForOffsetsParallel(
                            List.of(-1, -2, -3, -4, -5, -6, -7), 3);
                    long costMs = java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
                    taskLog.info("[ConversionBackfill] done: fetched={} matched={} sent={} failed={} skipped={} alreadySent={}",
                            r.fetched.get(), r.matched.get(), r.sent.get(), r.failed.get(),
                            r.skipped.get(), r.alreadySent.get());
                    taskLog.info("[ConversionBackfill] total cost={}ms", costMs);
                } catch (Exception e) {
                    long costMs = java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
                    taskLog.error("[ConversionBackfill] error after {}ms: {}", costMs, e.getMessage(), e);
                }
            };

            if (props.isSkipLeaderElection() || leaderElection == null) {
                job.run();
                return;
            }

            executor.submit(() ->
                    leaderElection.runOnce(
                            "adx:lock:tencent:conversion-backfill",
                            props.getTaskLockTtl(), props.getTaskLockRenewInterval(),
                            stopped::get,
                            stop -> job.run(),
                            e -> taskLog.error("tencent conversion backfill leader: {}", e.getMessage(), e)
                    )
            );
        }

        private void runCallbackRetry(LeaderElection.StopSignal jobStop) {
            long intervalMs = props.getCallbackRetryInterval().toMillis();
            taskLog.info("[CallbackRetry] task started, interval={}ms, limit={}", intervalMs, props.getCallbackRetryLimit());
            while (!jobStop.isStopped()) {
                try {
                    taskLog.info("[CallbackRetry] executing...");
                    RetryService.RetryResult r = retryService.retryTencentCallbacks(props.getCallbackRetryLimit());
                    taskLog.info("[CallbackRetry] done: fetched={} sent={} failed={}", r.fetched, r.sent, r.failed);
                } catch (Exception e) {
                    taskLog.error("[CallbackRetry] error: {}", e.getMessage(), e);
                }
                taskLog.info("[CallbackRetry] sleeping {}ms until next run", intervalMs);
                sleepResponsive(intervalMs, jobStop);
            }
            taskLog.info("[CallbackRetry] task stopped");
        }

        private void runTagEventSync(LeaderElection.StopSignal jobStop) {
            long intervalMs = props.getTagEventSyncInterval().toMillis();
            taskLog.info("[TagEventSync] task started, interval={}ms", intervalMs);
            while (!jobStop.isStopped()) {
                try {
                    int n = tagEventSyncService.syncOnce();
                    taskLog.info("[TagEventSync] done: synced={}", n);
                } catch (Exception e) {
                    taskLog.error("[TagEventSync] error: {}", e.getMessage(), e);
                }
                sleepResponsive(intervalMs, jobStop);
            }
            taskLog.info("[TagEventSync] task stopped");
        }

        private static void sleepResponsive(long ms, LeaderElection.StopSignal jobStop) {
            long deadline = System.currentTimeMillis() + ms;
            while (!jobStop.isStopped()) {
                long remaining = deadline - System.currentTimeMillis();
                if (remaining <= 0) break;
                try {
                    Thread.sleep(Math.min(remaining, 200));
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }

        private static void sleep(long ms) {
            try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        }
    }
}