tencent-baidu-tracking/src/main/java/com/adx/tencent/httpapi/AdminController.java

package com.adx.tencent.httpapi;

import com.adx.tencent.baidu.ConversionClient;
import com.adx.tencent.baidu.model.ConversionQuery;
import com.adx.tencent.conversionsync.ConversionBackfillJobService;
import com.adx.tencent.conversionsync.ConversionSyncRunner;
import com.adx.tencent.conversionsync.ConversionSyncService;
import com.adx.tencent.conversionsync.ManualTencentCallbackService;
import com.adx.tencent.conversionsync.RetryService;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.http.HttpStatus;
import org.springframework.lang.Nullable;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.server.ResponseStatusException;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * 管理接口：手动触发转化同步和回调重试。
 */
@RestController
@RequestMapping("/admin")
public class AdminController {

    private final ConversionClient conversionClient;
    private final ConversionSyncRunner conversionSyncRunner;
    private final ConversionBackfillJobService backfillJobService;
    private final ConversionSyncService conversionSyncer;
    private final RetryService retryService;
    private final ManualTencentCallbackService manualTencentCallbackService;
    private final StringRedisTemplate redisTemplate;

    public AdminController(ConversionClient conversionClient,
                           @Nullable ConversionSyncRunner conversionSyncRunner,
                           @Nullable ConversionBackfillJobService backfillJobService,
                           @Nullable ConversionSyncService conversionSyncer,
                           @Nullable RetryService retryService,
                           @Nullable ManualTencentCallbackService manualTencentCallbackService,
                           @Nullable StringRedisTemplate redisTemplate) {
        this.conversionClient = conversionClient;
        this.conversionSyncRunner = conversionSyncRunner;
        this.backfillJobService = backfillJobService;
        this.conversionSyncer = conversionSyncer;
        this.retryService = retryService;
        this.manualTencentCallbackService = manualTencentCallbackService;
        this.redisTemplate = redisTemplate;
    }

    @PostMapping("/conversions/query")
    public Object queryConversions(@RequestBody ConversionQuery query) throws Exception {
        return conversionClient.queryPayments(query);
    }

    @PostMapping("/conversions/sync")
    public Object syncConversions(@RequestBody ConversionQuery query) throws Exception {
        if (conversionSyncer == null) {
            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
                    "conversion syncer is not configured");
        }
        return conversionSyncer.syncTencentConversions(query);
    }

    @PostMapping("/conversions/backfill")
    public Object backfillConversions(@RequestBody(required = false) Map<String, Object> body) throws Exception {
        if (backfillJobService == null) {
            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
                    "conversion backfill job service is not configured");
        }
        List<Integer> offsets = parseOffsets(body);
        return backfillJobService.submit(offsets);
    }

    @GetMapping("/conversions/backfill/{jobId}")
    public Object getBackfillJob(@PathVariable("jobId") String jobId) {
        if (backfillJobService == null) {
            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
                    "conversion backfill job service is not configured");
        }
        ConversionBackfillJobService.JobState job = backfillJobService.getJob(jobId);
        if (job == null) {
            throw new ResponseStatusException(HttpStatus.NOT_FOUND, "job not found");
        }
        return job;
    }

    @PostMapping("/callbacks/retry")
    public Object retryCallbacks(@RequestBody Map<String, Object> body) {
        if (retryService == null) {
            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
                    "callback retryer is not configured");
        }
        int limit = toInt(body.get("limit"));
        return retryService.retryTencentCallbacks(limit);
    }

    @PostMapping("/callbacks/tencent/manual-replay")
    public Object manualReplayTencentCallback(@RequestBody Map<String, Object> body) {
        if (manualTencentCallbackService == null) {
            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
                    "manual tencent callback service is not configured");
        }
        long id = toLong(body.get("id"));
        if (id <= 0) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "id is required");
        }
        return manualTencentCallbackService.replayDeductedCallback(id);
    }

    @PostMapping("/redis/tighten-bid-media-ttl")
    public Object tightenBidMediaTtl(@RequestBody(required = false) Map<String, Object> body) {
        if (redisTemplate == null) {
            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
                    "redis template is not configured");
        }
        long ttlSeconds = toLong(body == null ? null : body.get("ttlSeconds"));
        if (ttlSeconds <= 0) {
            ttlSeconds = Duration.ofDays(3).getSeconds();
        }
        long scanCount = toLong(body == null ? null : body.get("scanCount"));
        if (scanCount <= 0) {
            scanCount = 2000;
        }
        long maxKeys = toLong(body == null ? null : body.get("maxKeys"));
        if (maxKeys <= 0) {
            maxKeys = 200000;
        }
        List<String> patterns = parseStringList(body == null ? null : body.get("patterns"));
        if (patterns.isEmpty()) {
            patterns = defaultBidMediaPatterns();
        }

        List<Map<String, Object>> results = new ArrayList<>();
        long totalScanned = 0;
        long totalUpdated = 0;
        for (String pattern : patterns) {
            TtlTightenResult result = tightenTtl(pattern, ttlSeconds, scanCount, Math.max(0, maxKeys - totalScanned));
            results.add(Map.of(
                    "pattern", pattern,
                    "scanned", result.scanned(),
                    "updated", result.updated(),
                    "stoppedByLimit", result.stoppedByLimit()
            ));
            totalScanned += result.scanned();
            totalUpdated += result.updated();
            if (totalScanned >= maxKeys) {
                break;
            }
        }
        return Map.of(
                "ttlSeconds", ttlSeconds,
                "ttlHuman", Duration.ofSeconds(ttlSeconds).toString(),
                "scanCount", scanCount,
                "maxKeys", maxKeys,
                "scanned", totalScanned,
                "updated", totalUpdated,
                "patterns", results
        );
    }

    @PostMapping("/redis/delete-bid-media-expiring-before")
    public Object deleteBidMediaExpiringBefore(@RequestBody(required = false) Map<String, Object> body) {
        if (redisTemplate == null) {
            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
                    "redis template is not configured");
        }
        long ttlLessThanSeconds = toLong(body == null ? null : body.get("ttlLessThanSeconds"));
        if (ttlLessThanSeconds <= 0) {
            ttlLessThanSeconds = Duration.ofDays(2).getSeconds();
        }
        long scanCount = toLong(body == null ? null : body.get("scanCount"));
        if (scanCount <= 0) {
            scanCount = 1000;
        }
        long maxKeys = toLong(body == null ? null : body.get("maxKeys"));
        if (maxKeys <= 0) {
            maxKeys = 50000;
        }
        List<String> patterns = parseStringList(body == null ? null : body.get("patterns"));
        if (patterns.isEmpty()) {
            patterns = defaultBidMediaPatterns();
        }

        String jobId = UUID.randomUUID().toString();
        RedisMaintenanceJob job = new RedisMaintenanceJob(jobId, "RUNNING", ttlLessThanSeconds, scanCount, maxKeys,
                0, 0, null, List.of());
        saveRedisMaintenanceJob(job);
        List<String> taskPatterns = List.copyOf(patterns);
        long taskTtlLessThanSeconds = ttlLessThanSeconds;
        long taskScanCount = scanCount;
        long taskMaxKeys = maxKeys;
        CompletableFuture.runAsync(() -> runDeleteExpiringJob(jobId, taskPatterns, taskTtlLessThanSeconds,
                taskScanCount, taskMaxKeys));
        return Map.of(
                "jobId", jobId,
                "status", "RUNNING",
                "ttlLessThanSeconds", ttlLessThanSeconds,
                "ttlLessThanHuman", Duration.ofSeconds(ttlLessThanSeconds).toString(),
                "scanCount", scanCount,
                "maxKeys", maxKeys,
                "patterns", taskPatterns
        );
    }

    @GetMapping("/redis/delete-bid-media-expiring-before/{jobId}")
    public Object getDeleteBidMediaExpiringBeforeJob(@PathVariable("jobId") String jobId) {
        Map<Object, Object> job = readRedisMaintenanceJob(jobId);
        if (job == null) {
            throw new ResponseStatusException(HttpStatus.NOT_FOUND, "job not found");
        }
        return job;
    }

    private void runDeleteExpiringJob(String jobId, List<String> patterns, long ttlLessThanSeconds,
                                      long scanCount, long maxKeys) {
        List<Map<String, Object>> results = new ArrayList<>();
        long totalScanned = 0;
        long totalDeleted = 0;
        try {
            for (String pattern : patterns) {
                DeleteExpiringResult result = deleteExpiringKeys(pattern, ttlLessThanSeconds, scanCount,
                        Math.max(0, maxKeys - totalScanned));
                results.add(Map.of(
                        "pattern", pattern,
                        "scanned", result.scanned(),
                        "deleted", result.deleted(),
                        "stoppedByLimit", result.stoppedByLimit()
                ));
                totalScanned += result.scanned();
                totalDeleted += result.deleted();
                saveRedisMaintenanceJob(new RedisMaintenanceJob(jobId, "RUNNING", ttlLessThanSeconds,
                        scanCount, maxKeys, totalScanned, totalDeleted, null, List.copyOf(results)));
                if (totalScanned >= maxKeys) {
                    break;
                }
            }
            saveRedisMaintenanceJob(new RedisMaintenanceJob(jobId, "COMPLETED", ttlLessThanSeconds,
                    scanCount, maxKeys, totalScanned, totalDeleted, null, List.copyOf(results)));
        } catch (Exception e) {
            saveRedisMaintenanceJob(new RedisMaintenanceJob(jobId, "FAILED", ttlLessThanSeconds,
                    scanCount, maxKeys, totalScanned, totalDeleted, rootMessage(e), List.copyOf(results)));
        }
    }

    @PostMapping("/redis/delete-bid-media-expiring-before-sync")
    public Object deleteBidMediaExpiringBeforeSync(@RequestBody(required = false) Map<String, Object> body) {
        if (redisTemplate == null) {
            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
                    "redis template is not configured");
        }
        long ttlLessThanSeconds = toLong(body == null ? null : body.get("ttlLessThanSeconds"));
        if (ttlLessThanSeconds <= 0) {
            ttlLessThanSeconds = Duration.ofDays(2).getSeconds();
        }
        long scanCount = toLong(body == null ? null : body.get("scanCount"));
        if (scanCount <= 0) {
            scanCount = 1000;
        }
        long maxKeys = toLong(body == null ? null : body.get("maxKeys"));
        if (maxKeys <= 0) {
            maxKeys = 50000;
        }
        List<String> patterns = parseStringList(body == null ? null : body.get("patterns"));
        if (patterns.isEmpty()) {
            patterns = defaultBidMediaPatterns();
        }

        List<Map<String, Object>> results = new ArrayList<>();
        long totalScanned = 0;
        long totalDeleted = 0;
        for (String pattern : patterns) {
            DeleteExpiringResult result = deleteExpiringKeys(pattern, ttlLessThanSeconds, scanCount,
                    Math.max(0, maxKeys - totalScanned));
            results.add(Map.of(
                    "pattern", pattern,
                    "scanned", result.scanned(),
                    "deleted", result.deleted(),
                    "stoppedByLimit", result.stoppedByLimit()
            ));
            totalScanned += result.scanned();
            totalDeleted += result.deleted();
            if (totalScanned >= maxKeys) {
                break;
            }
        }
        return Map.of(
                "ttlLessThanSeconds", ttlLessThanSeconds,
                "ttlLessThanHuman", Duration.ofSeconds(ttlLessThanSeconds).toString(),
                "scanCount", scanCount,
                "maxKeys", maxKeys,
                "scanned", totalScanned,
                "deleted", totalDeleted,
                "patterns", results
        );
    }

    private static int toInt(Object v) {
        if (v == null) return 0;
        if (v instanceof Number n) return n.intValue();
        try { return Integer.parseInt(v.toString()); } catch (NumberFormatException e) { return 0; }
    }

    private static long toLong(Object v) {
        if (v == null) return 0L;
        if (v instanceof Number n) return n.longValue();
        try { return Long.parseLong(v.toString()); } catch (NumberFormatException e) { return 0L; }
    }

    private TtlTightenResult tightenTtl(String pattern, long ttlSeconds, long scanCount, long maxKeys) {
        if (maxKeys <= 0 || pattern == null || pattern.isBlank()) {
            return new TtlTightenResult(0, 0, true);
        }
        long scanned = 0;
        long updated = 0;
        ScanOptions options = ScanOptions.scanOptions().match(pattern).count(scanCount).build();
        try (Cursor<String> cursor = redisTemplate.scan(options)) {
            while (cursor.hasNext()) {
                String key = cursor.next();
                scanned++;
                Long currentTtl = redisTemplate.getExpire(key, TimeUnit.SECONDS);
                if (currentTtl == null || currentTtl < 0 || currentTtl > ttlSeconds) {
                    Boolean ok = redisTemplate.expire(key, Duration.ofSeconds(ttlSeconds));
                    if (Boolean.TRUE.equals(ok)) {
                        updated++;
                    }
                }
                if (scanned >= maxKeys) {
                    return new TtlTightenResult(scanned, updated, true);
                }
            }
        }
        return new TtlTightenResult(scanned, updated, false);
    }

    private DeleteExpiringResult deleteExpiringKeys(String pattern, long ttlLessThanSeconds, long scanCount, long maxKeys) {
        if (maxKeys <= 0 || pattern == null || pattern.isBlank()) {
            return new DeleteExpiringResult(0, 0, true);
        }
        long scanned = 0;
        long deleted = 0;
        ScanOptions options = ScanOptions.scanOptions().match(pattern).count(scanCount).build();
        try (Cursor<String> cursor = redisTemplate.scan(options)) {
            while (cursor.hasNext()) {
                String key = cursor.next();
                scanned++;
                Long currentTtl = redisTemplate.getExpire(key, TimeUnit.SECONDS);
                if (currentTtl != null && currentTtl > 0 && currentTtl < ttlLessThanSeconds) {
                    Boolean ok = redisTemplate.delete(key);
                    if (Boolean.TRUE.equals(ok)) {
                        deleted++;
                    }
                }
                if (scanned >= maxKeys) {
                    return new DeleteExpiringResult(scanned, deleted, true);
                }
            }
        }
        return new DeleteExpiringResult(scanned, deleted, false);
    }

    private void saveRedisMaintenanceJob(RedisMaintenanceJob job) {
        try {
            String key = redisJobKey(job.jobId());
            Map<String, String> values = Map.of(
                    "jobId", job.jobId(),
                    "status", job.status(),
                    "ttlLessThanSeconds", String.valueOf(job.ttlLessThanSeconds()),
                    "scanCount", String.valueOf(job.scanCount()),
                    "maxKeys", String.valueOf(job.maxKeys()),
                    "scanned", String.valueOf(job.scanned()),
                    "deleted", String.valueOf(job.deleted()),
                    "error", job.error() == null ? "" : job.error(),
                    "patterns", job.patterns().toString()
            );
            redisTemplate.opsForHash().putAll(key, values);
            redisTemplate.expire(key, Duration.ofHours(2));
        } catch (Exception ignored) {
            // Job 状态只用于管理查询，写状态失败不影响清理任务继续执行。
        }
    }

    private Map<Object, Object> readRedisMaintenanceJob(String jobId) {
        Map<Object, Object> payload = redisTemplate.opsForHash().entries(redisJobKey(jobId));
        if (payload == null || payload.isEmpty()) {
            return null;
        }
        return payload;
    }

    private static List<String> parseStringList(Object raw) {
        if (!(raw instanceof Iterable<?> iterable)) {
            return List.of();
        }
        List<String> values = new ArrayList<>();
        for (Object item : iterable) {
            if (item != null && !item.toString().isBlank()) {
                values.add(item.toString());
            }
        }
        return values;
    }

    private static List<String> defaultBidMediaPatterns() {
        return List.of(
                "adx:bid:*",
                "adx:media:*",
                "adx:honor:bid:*",
                "adx:honor:media:*",
                "adx:kuaishou:bid:*",
                "adx:kuaishou:media:*",
                "adx:vivo:bid:*",
                "adx:vivo:media:*"
        );
    }

    private static List<Integer> parseOffsets(Map<String, Object> body) {
        if (body == null || body.isEmpty()) {
            return defaultBackfillOffsets();
        }
        Object raw = body.get("offsets");
        if (!(raw instanceof Iterable<?> iterable)) {
            return defaultBackfillOffsets();
        }
        List<Integer> offsets = new java.util.ArrayList<>();
        for (Object item : iterable) {
            offsets.add(toInt(item));
        }
        return offsets.isEmpty() ? defaultBackfillOffsets() : offsets;
    }

    private static List<Integer> defaultBackfillOffsets() {
        return List.of(-1, -2, -3, -4, -5, -6, -7);
    }

    private static String redisJobKey(String jobId) {
        return "adx:admin:redis-maintenance-job:" + jobId;
    }

    private static String rootMessage(Throwable e) {
        Throwable cur = e;
        while (cur.getCause() != null) {
            cur = cur.getCause();
        }
        return cur.getMessage() == null ? cur.getClass().getName() : cur.getMessage();
    }

    private record TtlTightenResult(long scanned, long updated, boolean stoppedByLimit) {}

    private record DeleteExpiringResult(long scanned, long deleted, boolean stoppedByLimit) {}

    private record RedisMaintenanceJob(String jobId, String status, long ttlLessThanSeconds, long scanCount,
                                       long maxKeys, long scanned, long deleted, String error,
                                       List<Map<String, Object>> patterns) {}
}