1

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

@@ -6,6 +6,7 @@ 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;
@@ -211,6 +212,12 @@ public class AppConfiguration {
     }
 
     @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());

src/main/java/com/adx/tencent/conversionsync/ConversionBackfillJobService.java

@@ -0,0 +1,67 @@
+package com.adx.tencent.conversionsync;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * 后台补拉任务服务。
+ * 用于把 backfill 从同步 HTTP 请求改成异步 job。
+ */
+public class ConversionBackfillJobService {
+
+    private static final Logger log = LoggerFactory.getLogger(ConversionBackfillJobService.class);
+
+    private final ConversionSyncRunner runner;
+    private final ExecutorService executor;
+    private final Map<String, JobState> jobs = new ConcurrentHashMap<>();
+
+    public ConversionBackfillJobService(ConversionSyncRunner runner) {
+        this.runner = runner;
+        this.executor = Executors.newFixedThreadPool(2, r -> {
+            Thread t = new Thread(r, "conversion-backfill-job");
+            t.setDaemon(true);
+            return t;
+        });
+    }
+
+    public JobHandle submit(List<Integer> offsets) {
+        String jobId = UUID.randomUUID().toString().replace("-", "");
+        JobState state = new JobState(jobId, "QUEUED", Instant.now(), null, null);
+        jobs.put(jobId, state);
+
+        executor.submit(() -> {
+            update(jobId, "RUNNING", null, null);
+            try {
+                ConversionSyncService.SyncResult result = runner.runForOffsetsParallel(offsets, 3);
+                update(jobId, "SUCCEEDED", result, null);
+            } catch (Exception e) {
+                log.error("[ConversionBackfill] job failed: jobId={}, error={}", jobId, e.getMessage(), e);
+                update(jobId, "FAILED", null, e.getMessage());
+            }
+        });
+
+        return new JobHandle(jobId, state.status(), state.createdAt());
+    }
+
+    public JobState getJob(String jobId) {
+        return jobs.get(jobId);
+    }
+
+    private void update(String jobId, String status, ConversionSyncService.SyncResult result, String error) {
+        JobState current = jobs.get(jobId);
+        if (current == null) return;
+        jobs.put(jobId, new JobState(jobId, status, current.createdAt(), result, error));
+    }
+
+    public record JobHandle(String jobId, String status, Instant createdAt) {}
+
+    public record JobState(String jobId, String status, Instant createdAt,
+                           ConversionSyncService.SyncResult result, String errorMessage) {}
+}

src/main/java/com/adx/tencent/conversionsync/ConversionSyncRunner.java

@@ -8,6 +8,7 @@ import java.time.LocalDate;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.*;
 
 /**
  * 转化同步调度器。
@@ -22,6 +23,7 @@ public class ConversionSyncRunner {
     private final int dateOffsetDays;
     private final int pageSize;
     private final List<Integer> acts;
+    private final ExecutorService dateExecutor;
 
     public ConversionSyncRunner(ConversionSyncService syncer,
                                 int dateOffsetDays, int pageSize, List<Integer> acts) {
@@ -29,6 +31,11 @@ public class ConversionSyncRunner {
         this.dateOffsetDays = dateOffsetDays;
         this.pageSize = pageSize > 0 ? pageSize : 1;
         this.acts = acts;
+        this.dateExecutor = Executors.newFixedThreadPool(3, r -> {
+            Thread t = new Thread(r, "conv-backfill-date");
+            t.setDaemon(true);
+            return t;
+        });
     }
 
     public ConversionSyncService.SyncResult runOnce() throws Exception {
@@ -36,6 +43,10 @@ public class ConversionSyncRunner {
     }
 
     public ConversionSyncService.SyncResult runForOffsets(List<Integer> offsets) throws Exception {
+        return runForOffsetsParallel(offsets, 1);
+    }
+
+    public ConversionSyncService.SyncResult runForOffsetsParallel(List<Integer> offsets, int maxConcurrency) throws Exception {
         ConversionSyncService.SyncResult total = new ConversionSyncService.SyncResult();
         if (offsets == null || offsets.isEmpty()) {
             return total;
@@ -48,10 +59,32 @@ public class ConversionSyncRunner {
             }
         }
 
+        int concurrency = Math.max(1, Math.min(maxConcurrency, normalized.size()));
+        ExecutorService executor = concurrency == 1 ? null : this.dateExecutor;
+        if (executor == null) {
+            for (Integer offset : normalized) {
+                String date = LocalDate.now().plusDays(offset).format(DATE_FMT);
+                ConversionSyncService.SyncResult result = runForDate(date);
+                merge(total, result);
+            }
+            return total;
+        }
+
+        List<CompletableFuture<ConversionSyncService.SyncResult>> futures = new ArrayList<>(normalized.size());
         for (Integer offset : normalized) {
             String date = LocalDate.now().plusDays(offset).format(DATE_FMT);
-            ConversionSyncService.SyncResult result = runForDate(date);
-            merge(total, result);
+            futures.add(CompletableFuture.supplyAsync(() -> {
+                try {
+                    return runForDate(date);
+                } catch (Exception e) {
+                    throw new CompletionException(e);
+                }
+            }, executor));
+        }
+
+        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+        for (CompletableFuture<ConversionSyncService.SyncResult> future : futures) {
+            merge(total, future.join());
         }
         return total;
     }

src/main/java/com/adx/tencent/conversionsync/ConversionSyncService.java

@@ -198,7 +198,7 @@ public class ConversionSyncService {
 
     private static ConversionRecord conversionRecord(PaymentInfo p, BidRecord bid, String media) {
         ConversionRecord r = new ConversionRecord();
-        r.setDedupeKey(conversionDedupeKey(p, media));
+        r.setDedupeKey(conversionDedupeKey(p));
         r.setQk(p.getQk());
         r.setMedia(media);
         r.setTagId(bid != null ? bid.getTagId() : null);
@@ -216,8 +216,8 @@ public class ConversionSyncService {
         return r;
     }
 
-    static String conversionDedupeKey(PaymentInfo p, String media) {
-        return String.format("%s:%s:%s:%s:%d", media, p.getQk(), p.getDate(), p.getDeviceId(), p.getAct());
+    static String conversionDedupeKey(PaymentInfo p) {
+        return String.format("%s:%s:%s:%d", p.getQk(), p.getDate(), p.getDeviceId(), p.getAct());
     }
 
     static String callbackDedupeKey(PaymentInfo p, BidRecord bid, String actionType) {
@@ -232,7 +232,8 @@ public class ConversionSyncService {
     }
 
     static String legacyCallbackDedupeKey(PaymentInfo p, String media) {
-        return conversionDedupeKey(p, media) + ":" + p.getAct();
+        return String.format("%s:%s:%s:%s:%d:%d",
+                media, p.getQk(), p.getDate(), p.getDeviceId(), p.getAct(), p.getAct());
     }
 
     // ─── Result DTO ─────────────────────────────────────────────────────────

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

@@ -2,6 +2,7 @@ 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.RetryService;
@@ -22,15 +23,18 @@ public class AdminController {
 
     private final ConversionClient conversionClient;
     private final ConversionSyncRunner conversionSyncRunner;
+    private final ConversionBackfillJobService backfillJobService;
     private final ConversionSyncService conversionSyncer;
     private final RetryService retryService;
 
     public AdminController(ConversionClient conversionClient,
                            @Nullable ConversionSyncRunner conversionSyncRunner,
+                           @Nullable ConversionBackfillJobService backfillJobService,
                            @Nullable ConversionSyncService conversionSyncer,
                            @Nullable RetryService retryService) {
         this.conversionClient = conversionClient;
         this.conversionSyncRunner = conversionSyncRunner;
+        this.backfillJobService = backfillJobService;
         this.conversionSyncer = conversionSyncer;
         this.retryService = retryService;
     }
@@ -51,12 +55,25 @@ public class AdminController {
 
     @PostMapping("/conversions/backfill")
     public Object backfillConversions(@RequestBody(required = false) Map<String, Object> body) throws Exception {
-        if (conversionSyncRunner == null) {
+        if (backfillJobService == null) {
             throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
-                    "conversion sync runner is not configured");
+                    "conversion backfill job service is not configured");
         }
         List<Integer> offsets = parseOffsets(body);
-        return conversionSyncRunner.runForOffsets(offsets);
+        return backfillJobService.submit(offsets);
+    }
+
+    @GetMapping("/conversions/backfill/{jobId}")
+    public Object getBackfillJob(@PathVariable 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")

src/main/java/com/adx/tencent/storage/typehandler/JsonStringListTypeHandler.java

@@ -0,0 +1,52 @@
+package com.adx.tencent.storage.typehandler;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.ibatis.type.BaseTypeHandler;
+import org.apache.ibatis.type.JdbcType;
+
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.List;
+
+public class JsonStringListTypeHandler extends BaseTypeHandler<List<String>> {
+
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+    private static final TypeReference<List<String>> TYPE = new TypeReference<>() {};
+
+    @Override
+    public void setNonNullParameter(PreparedStatement ps, int i, List<String> parameter, JdbcType jdbcType) throws SQLException {
+        try {
+            ps.setString(i, MAPPER.writeValueAsString(parameter));
+        } catch (Exception e) {
+            throw new SQLException("serialize list json failed", e);
+        }
+    }
+
+    @Override
+    public List<String> getNullableResult(ResultSet rs, String columnName) throws SQLException {
+        return parse(rs.getString(columnName));
+    }
+
+    @Override
+    public List<String> getNullableResult(ResultSet rs, int columnIndex) throws SQLException {
+        return parse(rs.getString(columnIndex));
+    }
+
+    @Override
+    public List<String> getNullableResult(CallableStatement cs, int columnIndex) throws SQLException {
+        return parse(cs.getString(columnIndex));
+    }
+
+    private List<String> parse(String json) throws SQLException {
+        if (json == null || json.isBlank()) {
+            return null;
+        }
+        try {
+            List<String> values = MAPPER.readValue(json, TYPE);
+            return values != null ? values : new ArrayList<>();
+        } catch (Exception e) {
+            throw new SQLException("parse list json failed", e);
+        }
+    }
+}

src/main/java/com/adx/tencent/storage/typehandler/JsonStringMapTypeHandler.java

@@ -0,0 +1,52 @@
+package com.adx.tencent.storage.typehandler;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.ibatis.type.BaseTypeHandler;
+import org.apache.ibatis.type.JdbcType;
+
+import java.sql.*;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public class JsonStringMapTypeHandler extends BaseTypeHandler<Map<String, String>> {
+
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+    private static final TypeReference<Map<String, String>> TYPE = new TypeReference<>() {};
+
+    @Override
+    public void setNonNullParameter(PreparedStatement ps, int i, Map<String, String> parameter, JdbcType jdbcType) throws SQLException {
+        try {
+            ps.setString(i, MAPPER.writeValueAsString(parameter));
+        } catch (Exception e) {
+            throw new SQLException("serialize map json failed", e);
+        }
+    }
+
+    @Override
+    public Map<String, String> getNullableResult(ResultSet rs, String columnName) throws SQLException {
+        return parse(rs.getString(columnName));
+    }
+
+    @Override
+    public Map<String, String> getNullableResult(ResultSet rs, int columnIndex) throws SQLException {
+        return parse(rs.getString(columnIndex));
+    }
+
+    @Override
+    public Map<String, String> getNullableResult(CallableStatement cs, int columnIndex) throws SQLException {
+        return parse(cs.getString(columnIndex));
+    }
+
+    private Map<String, String> parse(String json) throws SQLException {
+        if (json == null || json.isBlank()) {
+            return null;
+        }
+        try {
+            Map<String, String> values = MAPPER.readValue(json, TYPE);
+            return values != null ? values : new LinkedHashMap<>();
+        } catch (Exception e) {
+            throw new SQLException("parse map json failed", e);
+        }
+    }
+}

src/main/resources/mapper/BidEventMapper.xml

@@ -16,12 +16,12 @@
         <result property="impId" column="imp_id"/>
         <result property="tagId" column="tag_id"/>
         <result property="price" column="price"/>
-        <result property="showUrls" column="show_urls" typeHandler="org.apache.ibatis.type.JacksonTypeHandler"/>
-        <result property="clickUrls" column="click_urls" typeHandler="org.apache.ibatis.type.JacksonTypeHandler"/>
+        <result property="showUrls" column="show_urls" typeHandler="com.adx.tencent.storage.typehandler.JsonStringListTypeHandler"/>
+        <result property="clickUrls" column="click_urls" typeHandler="com.adx.tencent.storage.typehandler.JsonStringListTypeHandler"/>
         <result property="landingPage" column="landing_page"/>
         <result property="appStoreLink" column="app_store_link"/>
         <result property="packageName" column="package_name"/>
-        <result property="mediaParams" column="media_params" typeHandler="org.apache.ibatis.type.JacksonTypeHandler"/>
+        <result property="mediaParams" column="media_params" typeHandler="com.adx.tencent.storage.typehandler.JsonStringMapTypeHandler"/>
         <result property="createdAt" column="created_at"/>
     </resultMap>
 

src/main/resources/mapper/ConversionMapper.xml

@@ -11,8 +11,14 @@
             #{deviceId}, #{conv}, #{payment}, #{gmv}, #{act}, #{tu}, #{clkTime}, #{createdAt}
         )
         ON DUPLICATE KEY UPDATE
-            media = VALUES(media),
-            tag_id = VALUES(tag_id),
+            media = CASE
+                WHEN VALUES(media) IS NOT NULL AND VALUES(media) != '' THEN VALUES(media)
+                ELSE media
+            END,
+            tag_id = CASE
+                WHEN VALUES(tag_id) IS NOT NULL AND VALUES(tag_id) != '' THEN VALUES(tag_id)
+                ELSE tag_id
+            END,
             conv = VALUES(conv),
             payment = VALUES(payment),
             gmv = VALUES(gmv),