腾讯配置

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

@@ -461,9 +461,10 @@ public class AppConfiguration {
 
     @Bean
     public VivoRetryService vivoRetryService(@Nullable VivoColdStore vivoColdStore,
-                                             @Nullable VivoClient vivoClient) {
+                                             @Nullable VivoClient vivoClient,
+                                             ObjectMapper objectMapper) {
         if (vivoColdStore == null || vivoClient == null) return null;
-        return new VivoRetryService(vivoColdStore, vivoClient, props.getVivoCallbackRetryLimit());
+        return new VivoRetryService(vivoColdStore, vivoClient, objectMapper, props.getVivoCallbackRetryLimit());
     }
 
     @Bean

src/main/java/com/adx/tencent/httpapi/AdminController.java

@@ -7,6 +7,7 @@ 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 com.adx.tencent.vivo.service.VivoRetryService;
 import org.springframework.http.HttpStatus;
 import org.springframework.lang.Nullable;
 import org.springframework.web.bind.annotation.*;
@@ -28,19 +29,22 @@ public class AdminController {
     private final ConversionSyncService conversionSyncer;
     private final RetryService retryService;
     private final ManualTencentCallbackService manualTencentCallbackService;
+    private final VivoRetryService vivoRetryService;
 
     public AdminController(ConversionClient conversionClient,
                            @Nullable ConversionSyncRunner conversionSyncRunner,
                            @Nullable ConversionBackfillJobService backfillJobService,
                            @Nullable ConversionSyncService conversionSyncer,
                            @Nullable RetryService retryService,
-                           @Nullable ManualTencentCallbackService manualTencentCallbackService) {
+                           @Nullable ManualTencentCallbackService manualTencentCallbackService,
+                           @Nullable VivoRetryService vivoRetryService) {
         this.conversionClient = conversionClient;
         this.conversionSyncRunner = conversionSyncRunner;
         this.backfillJobService = backfillJobService;
         this.conversionSyncer = conversionSyncer;
         this.retryService = retryService;
         this.manualTencentCallbackService = manualTencentCallbackService;
+        this.vivoRetryService = vivoRetryService;
     }
 
     @PostMapping("/conversions/query")
@@ -103,6 +107,18 @@ public class AdminController {
         return manualTencentCallbackService.replayDeductedCallback(id);
     }
 
+    @PostMapping("/callbacks/vivo/replay-src-id")
+    public Object replayVivoCallbacksReplacingSrcId(@RequestBody(required = false) Map<String, Object> body) {
+        if (vivoRetryService == null) {
+            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
+                    "vivo retry service is not configured");
+        }
+        String oldSrcId = toStringValue(body != null ? body.get("oldSrcId") : null, "ds-202608201718");
+        String newSrcId = toStringValue(body != null ? body.get("newSrcId") : null, "ds-202608253668");
+        int limit = toInt(body != null ? body.get("limit") : null);
+        return vivoRetryService.replayFailedCallbacksReplacingSrcId(oldSrcId, newSrcId, limit);
+    }
+
     private static int toInt(Object v) {
         if (v == null) return 0;
         if (v instanceof Number n) return n.intValue();
@@ -115,6 +131,12 @@ public class AdminController {
         try { return Long.parseLong(v.toString()); } catch (NumberFormatException e) { return 0L; }
     }
 
+    private static String toStringValue(Object v, String defaultValue) {
+        if (v == null) return defaultValue;
+        String s = v.toString();
+        return s == null || s.isBlank() ? defaultValue : s.trim();
+    }
+
     private static List<Integer> parseOffsets(Map<String, Object> body) {
         if (body == null || body.isEmpty()) {
             return defaultBackfillOffsets();

src/main/java/com/adx/tencent/vivo/client/VivoClient.java

@@ -110,6 +110,36 @@ public class VivoClient {
         return retry;
     }
 
+    public VivoMediaCallbackRecord replayCallback(VivoMediaCallbackRecord original, String requestBody) throws Exception {
+        VivoMediaCallbackRecord replay = copy(original);
+        replay.setAttempt(original.getAttempt() + 1);
+        replay.setCreatedAt(Instant.now());
+        replay.setRequestBody(requestBody);
+
+        String advertiserId = queryParam(original.getCallbackUrl(), "advertiser_id");
+        if (advertiserId == null || advertiserId.isBlank()) {
+            throw new IllegalStateException("missing advertiser_id in vivo callback url");
+        }
+        String accessToken = authService.getAccessTokenForAdvertiser(advertiserId);
+        long timestamp = System.currentTimeMillis();
+        String nonce = UUID.randomUUID().toString().replace("-", "");
+        String callbackUrl = buildCallbackUrl(advertiserId, accessToken, timestamp, nonce);
+
+        replay.setCallbackUrl(callbackUrl);
+        log.info("[Vivo] callback手动重放请求: qk={} tagId={} eventType={} attempt={} url={} body={}",
+                original.getQk(), original.getTagId(), original.getEventType(),
+                replay.getAttempt(), callbackUrl, requestBody);
+
+        HttpResponse<String> response = sendPost(callbackUrl, requestBody);
+        replay.setStatus(response.statusCode());
+        replay.setResponseBody(response.body());
+        applyResponse(replay, response.statusCode(), response.body());
+        log.info("[Vivo] callback手动重放结果: qk={} tagId={} eventType={} attempt={} status={} ok={} response={}",
+                replay.getQk(), replay.getTagId(), replay.getEventType(), replay.getAttempt(),
+                replay.getStatus(), replay.isOk(), replay.getResponseBody());
+        return replay;
+    }
+
     private VivoMediaCallbackRecord buildBaseRecord(VivoBidRecord bid,
                                                     PaymentInfo payment,
                                                     VivoTagEventRecord tagEvent,
@@ -298,6 +328,7 @@ public class VivoClient {
 
     private static VivoMediaCallbackRecord copy(VivoMediaCallbackRecord src) {
         VivoMediaCallbackRecord copy = new VivoMediaCallbackRecord();
+        copy.setId(src.getId());
         copy.setDedupeKey(src.getDedupeKey());
         copy.setQk(src.getQk());
         copy.setTagId(src.getTagId());

src/main/java/com/adx/tencent/vivo/mapper/VivoMediaCallbackMapper.java

@@ -11,4 +11,6 @@ public interface VivoMediaCallbackMapper {
     void batchInsert(@Param("records") List<VivoMediaCallbackRecord> records);
     List<String> selectSuccessfulDedupeKeys(@Param("dedupeKeys") List<String> dedupeKeys);
     List<VivoMediaCallbackRecord> selectPendingCallbacks(@Param("limit") int limit);
+    List<VivoMediaCallbackRecord> selectFailedCallbacksBySrcId(@Param("srcId") String srcId, @Param("limit") int limit);
+    void updateReplayResult(@Param("record") VivoMediaCallbackRecord record);
 }

src/main/java/com/adx/tencent/vivo/model/VivoMediaCallbackRecord.java

@@ -6,6 +6,7 @@ public class VivoMediaCallbackRecord {
     public static final String DISPATCH_STATUS_SENT = "SENT";
     public static final String DISPATCH_STATUS_DEDUCTED = "DEDUCTED";
 
+    private Long id;
     private String dedupeKey;
     private String qk;
     private String tagId;
@@ -24,6 +25,8 @@ public class VivoMediaCallbackRecord {
     private String requestBody;
     private Instant createdAt;
 
+    public Long getId() { return id; }
+    public void setId(Long v) { this.id = v; }
     public String getDedupeKey() { return dedupeKey; }
     public void setDedupeKey(String v) { this.dedupeKey = v; }
     public String getQk() { return qk; }

src/main/java/com/adx/tencent/vivo/service/VivoRetryService.java

@@ -3,6 +3,9 @@ package com.adx.tencent.vivo.service;
 import com.adx.tencent.vivo.client.VivoClient;
 import com.adx.tencent.vivo.model.VivoMediaCallbackRecord;
 import com.adx.tencent.vivo.store.VivoColdStore;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 
 import java.time.Instant;
 import java.util.ArrayList;
@@ -12,11 +15,13 @@ public class VivoRetryService {
 
     private final VivoColdStore coldStore;
     private final VivoClient vivoClient;
+    private final ObjectMapper objectMapper;
     private final int defaultLimit;
 
-    public VivoRetryService(VivoColdStore coldStore, VivoClient vivoClient, int defaultLimit) {
+    public VivoRetryService(VivoColdStore coldStore, VivoClient vivoClient, ObjectMapper objectMapper, int defaultLimit) {
         this.coldStore = coldStore;
         this.vivoClient = vivoClient;
+        this.objectMapper = objectMapper;
         this.defaultLimit = defaultLimit > 0 ? defaultLimit : 100;
     }
 
@@ -45,8 +50,70 @@ public class VivoRetryService {
         return result;
     }
 
+    public ReplaySrcIdResult replayFailedCallbacksReplacingSrcId(String oldSrcId, String newSrcId, int limit) {
+        String normalizedOld = requireSrcId(oldSrcId, "oldSrcId");
+        String normalizedNew = requireSrcId(newSrcId, "newSrcId");
+        List<VivoMediaCallbackRecord> pending = coldStore.failedMediaCallbacksBySrcId(
+                normalizedOld, limit <= 0 ? defaultLimit : limit);
+
+        ReplaySrcIdResult result = new ReplaySrcIdResult();
+        result.oldSrcId = normalizedOld;
+        result.newSrcId = normalizedNew;
+        result.fetched = pending.size();
+        for (VivoMediaCallbackRecord record : pending) {
+            String replacedBody = null;
+            try {
+                replacedBody = replaceSrcId(record.getRequestBody(), normalizedOld, normalizedNew);
+                if (replacedBody == null) {
+                    result.skipped++;
+                    continue;
+                }
+                VivoMediaCallbackRecord replay = vivoClient.replayCallback(record, replacedBody);
+                coldStore.updateMediaCallbackReplayResult(replay);
+                if (replay.isOk()) result.sent++;
+                else result.failed++;
+            } catch (Exception e) {
+                VivoMediaCallbackRecord failed = copy(record);
+                failed.setAttempt(record.getAttempt() + 1);
+                if (replacedBody != null) {
+                    failed.setRequestBody(replacedBody);
+                }
+                failed.setOk(false);
+                failed.setErrorMessage(e.getMessage());
+                failed.setCreatedAt(Instant.now());
+                coldStore.updateMediaCallbackReplayResult(failed);
+                result.failed++;
+            }
+        }
+        return result;
+    }
+
+    private String replaceSrcId(String requestBody, String oldSrcId, String newSrcId) throws Exception {
+        if (requestBody == null || requestBody.isBlank()) {
+            return null;
+        }
+        JsonNode root = objectMapper.readTree(requestBody);
+        if (!(root instanceof ObjectNode objectNode)) {
+            return null;
+        }
+        JsonNode srcId = objectNode.get("srcId");
+        if (srcId == null || !oldSrcId.equals(srcId.asText())) {
+            return null;
+        }
+        objectNode.put("srcId", newSrcId);
+        return objectMapper.writeValueAsString(objectNode);
+    }
+
+    private static String requireSrcId(String value, String field) {
+        if (value == null || value.isBlank()) {
+            throw new IllegalArgumentException(field + " is required");
+        }
+        return value.trim();
+    }
+
     private static VivoMediaCallbackRecord copy(VivoMediaCallbackRecord src) {
         VivoMediaCallbackRecord copy = new VivoMediaCallbackRecord();
+        copy.setId(src.getId());
         copy.setDedupeKey(src.getDedupeKey());
         copy.setQk(src.getQk());
         copy.setTagId(src.getTagId());
@@ -72,4 +139,13 @@ public class VivoRetryService {
         public int sent;
         public int failed;
     }
+
+    public static class ReplaySrcIdResult {
+        public String oldSrcId;
+        public String newSrcId;
+        public int fetched;
+        public int sent;
+        public int failed;
+        public int skipped;
+    }
 }

src/main/java/com/adx/tencent/vivo/store/VivoColdStore.java

@@ -256,6 +256,24 @@ public class VivoColdStore {
         }
     }
 
+    public List<VivoMediaCallbackRecord> failedMediaCallbacksBySrcId(String srcId, int limit) {
+        if (srcId == null || srcId.isBlank()) return List.of();
+        if (limit <= 0) limit = 100;
+        try (SqlSession session = sqlSessionFactory.openSession(true)) {
+            return session.getMapper(VivoMediaCallbackMapper.class).selectFailedCallbacksBySrcId(srcId.trim(), limit);
+        }
+    }
+
+    public void updateMediaCallbackReplayResult(VivoMediaCallbackRecord record) {
+        if (record == null || record.getId() == null || record.getId() <= 0) {
+            throw new IllegalArgumentException("vivo callback id is required");
+        }
+        if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
+        try (SqlSession session = sqlSessionFactory.openSession(true)) {
+            session.getMapper(VivoMediaCallbackMapper.class).updateReplayResult(record);
+        }
+    }
+
     public List<VivoTagEventRecord> getTagEvents(String tagId, int baiduAct) {
         if (tagId == null || tagId.isBlank()) return List.of();
         try (SqlSession session = sqlSessionFactory.openSession(true)) {

src/main/resources/mapper/VivoMediaCallbackMapper.xml

@@ -3,6 +3,7 @@
         "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="com.adx.tencent.vivo.mapper.VivoMediaCallbackMapper">
     <resultMap id="vivoMediaCallbackResultMap" type="com.adx.tencent.vivo.model.VivoMediaCallbackRecord">
+        <id property="id" column="id"/>
         <result property="dedupeKey" column="dedupe_key"/>
         <result property="qk" column="qk"/>
         <result property="tagId" column="tag_id"/>
@@ -47,7 +48,7 @@
     </select>
 
     <select id="selectPendingCallbacks" resultMap="vivoMediaCallbackResultMap">
-        SELECT m.dedupe_key, m.qk, m.tag_id, m.callback_url, m.event_type, m.event_time_ms,
+        SELECT m.id, m.dedupe_key, m.qk, m.tag_id, m.callback_url, m.event_type, m.event_time_ms,
                m.purchase, m.deduction_rate, m.is_direct_match, m.dispatch_status, m.status, m.ok AS ok_val, m.attempt, m.response_body, m.error_message,
                m.request_body, m.created_at
         FROM vivo_media_callbacks m
@@ -66,4 +67,30 @@
         ORDER BY m.created_at ASC
         LIMIT #{limit}
     </select>
+
+    <select id="selectFailedCallbacksBySrcId" resultMap="vivoMediaCallbackResultMap">
+        SELECT m.id, m.dedupe_key, m.qk, m.tag_id, m.callback_url, m.event_type, m.event_time_ms,
+               m.purchase, m.deduction_rate, m.is_direct_match, m.dispatch_status, m.status, m.ok AS ok_val, m.attempt, m.response_body, m.error_message,
+               m.request_body, m.created_at
+        FROM vivo_media_callbacks m
+        WHERE m.ok = 0
+          AND m.dispatch_status != 'DEDUCTED'
+          AND m.request_body LIKE CONCAT('%', #{srcId}, '%')
+        ORDER BY m.created_at ASC
+        LIMIT #{limit}
+    </select>
+
+    <update id="updateReplayResult">
+        UPDATE vivo_media_callbacks
+        SET callback_url = #{record.callbackUrl},
+            event_time_ms = #{record.eventTimeMs},
+            status = #{record.status},
+            ok = #{record.ok},
+            attempt = #{record.attempt},
+            response_body = #{record.responseBody},
+            error_message = #{record.errorMessage},
+            request_body = #{record.requestBody},
+            created_at = #{record.createdAtTs}
+        WHERE id = #{record.id}
+    </update>
 </mapper>