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 body) throws Exception { if (backfillJobService == null) { throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "conversion backfill job service is not configured"); } List 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 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 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 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 patterns = parseStringList(body == null ? null : body.get("patterns")); if (patterns.isEmpty()) { patterns = defaultBidMediaPatterns(); } List> 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 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 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 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 job = readRedisMaintenanceJob(jobId); if (job == null) { throw new ResponseStatusException(HttpStatus.NOT_FOUND, "job not found"); } return job; } private void runDeleteExpiringJob(String jobId, List patterns, long ttlLessThanSeconds, long scanCount, long maxKeys) { List> 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 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 patterns = parseStringList(body == null ? null : body.get("patterns")); if (patterns.isEmpty()) { patterns = defaultBidMediaPatterns(); } List> 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 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 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 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 readRedisMaintenanceJob(String jobId) { Map payload = redisTemplate.opsForHash().entries(redisJobKey(jobId)); if (payload == null || payload.isEmpty()) { return null; } return payload; } private static List parseStringList(Object raw) { if (!(raw instanceof Iterable iterable)) { return List.of(); } List values = new ArrayList<>(); for (Object item : iterable) { if (item != null && !item.toString().isBlank()) { values.add(item.toString()); } } return values; } private static List 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 parseOffsets(Map body) { if (body == null || body.isEmpty()) { return defaultBackfillOffsets(); } Object raw = body.get("offsets"); if (!(raw instanceof Iterable iterable)) { return defaultBackfillOffsets(); } List offsets = new java.util.ArrayList<>(); for (Object item : iterable) { offsets.add(toInt(item)); } return offsets.isEmpty() ? defaultBackfillOffsets() : offsets; } private static List 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> patterns) {} }