数据清理

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

@@ -456,7 +456,7 @@ public class ConversionSyncService {
             }
 
             if (validOppo) {
-                conversionRecords.add(toConversionRecord(payment, null, "oppo"));
+                conversionRecords.add(toOppoConversionRecord(payment, oppoBid));
                 result.matched.incrementAndGet();
                 CallbackTask oppoTask = buildOppoCallbackTask(payment, oppoBid);
                 if (oppoTask != null) callbackTasks.add(oppoTask);
@@ -1310,6 +1310,26 @@ public class ConversionSyncService {
         return record;
     }
 
+    static ConversionRecord toOppoConversionRecord(PaymentInfo payment, OppoBidRecord bid) {
+        ConversionRecord record = new ConversionRecord();
+        record.setDedupeKey(conversionDedupeKey(payment));
+        record.setQk(payment.getQk());
+        record.setMedia("oppo");
+        record.setTagId(bid != null ? bid.getTagId() : null);
+        record.setDate(payment.getDate());
+        record.setAppSid(payment.getAppSid());
+        record.setCustomerName(payment.getCustomerName());
+        record.setDeviceId(payment.getDeviceId());
+        record.setConv(payment.getConv());
+        record.setPayment(payment.getPayment());
+        record.setGmv(payment.getGmv());
+        record.setAct(payment.getAct());
+        record.setTu(payment.getTu());
+        record.setClkTime(payment.getClkTime());
+        record.setCreatedAt(Instant.now());
+        return record;
+    }
+
     static String conversionDedupeKey(PaymentInfo payment) {
         return String.format("%s:%s:%s:%d",
                 payment.getQk(),

src/main/java/com/adx/tencent/honor/store/HonorHotStore.java

@@ -183,7 +183,14 @@ public class HonorHotStore {
             String payload = payloads != null ? payloads.get(i) : null;
             String qk = keyToQk.get(keys.get(i));
             if (payload != null) {
-                result.put(qk, parseBidRecord(payload));
+                HonorBidRecord bid = parseBidRecord(payload);
+                if (isCompleteAttribution(bid)) {
+                    result.put(qk, bid);
+                } else if (qk != null) {
+                    missing.add(qk);
+                    log.warn("[HonorHotStore] incomplete cached bid, falling back to TiDB | qk={} | media={} | tagId={}",
+                            qk, bid == null ? null : bid.getMedia(), bid == null ? null : bid.getTagId());
+                }
             } else if (qk != null) {
                 missing.add(qk);
             }
@@ -196,6 +203,13 @@ public class HonorHotStore {
         return result;
     }
 
+    private static boolean isCompleteAttribution(HonorBidRecord bid) {
+        return bid != null
+                && "honor".equals(bid.getMedia())
+                && bid.getTagId() != null
+                && !bid.getTagId().isBlank();
+    }
+
     public boolean markConversionSeen(String dedupeKey) {
         if (dedupeKey == null || dedupeKey.isBlank()) {
             return false;

src/main/java/com/adx/tencent/kuaishou/store/KuaishouHotStore.java

@@ -175,7 +175,17 @@ public class KuaishouHotStore {
             String payload = payloads != null ? payloads.get(i) : null;
             String qk = keyToQk.get(keys.get(i));
             if (payload != null) {
-                result.put(qk, parseBidRecord(payload));
+                KuaishouBidRecord bid = parseBidRecord(payload);
+                // Older/slim Redis payloads can exist without attribution fields. They are
+                // sufficient for neither conversion attribution nor callback routing, so use
+                // the complete cold record instead of treating the cache key as a hit.
+                if (isCompleteAttribution(bid)) {
+                    result.put(qk, bid);
+                } else if (qk != null) {
+                    missing.add(qk);
+                    log.warn("[KuaishouHotStore] incomplete cached bid, falling back to TiDB | qk={} | media={} | tagId={}",
+                            qk, bid == null ? null : bid.getMedia(), bid == null ? null : bid.getTagId());
+                }
             } else if (qk != null) {
                 missing.add(qk);
             }
@@ -188,6 +198,13 @@ public class KuaishouHotStore {
         return result;
     }
 
+    private static boolean isCompleteAttribution(KuaishouBidRecord bid) {
+        return bid != null
+                && "kuaishou".equals(bid.getMedia())
+                && bid.getTagId() != null
+                && !bid.getTagId().isBlank();
+    }
+
     public boolean markConversionSeen(String dedupeKey) {
         if (dedupeKey == null || dedupeKey.isBlank()) {
             return false;

src/main/java/com/adx/tencent/oppo/store/OppoHotStore.java

@@ -156,7 +156,14 @@ public class OppoHotStore {
             String payload = payloads != null ? payloads.get(i) : null;
             String qk = keyToQk.get(keys.get(i));
             if (payload != null) {
-                result.put(qk, parseBidRecord(payload));
+                OppoBidRecord bid = parseBidRecord(payload);
+                if (isCompleteAttribution(bid)) {
+                    result.put(qk, bid);
+                } else if (qk != null) {
+                    missing.add(qk);
+                    log.warn("[OppoHotStore] incomplete cached bid, falling back to TiDB | qk={} | media={} | tagId={}",
+                            qk, bid == null ? null : bid.getMedia(), bid == null ? null : bid.getTagId());
+                }
             } else if (qk != null) {
                 missing.add(qk);
             }
@@ -169,6 +176,13 @@ public class OppoHotStore {
         return result;
     }
 
+    private static boolean isCompleteAttribution(OppoBidRecord bid) {
+        return bid != null
+                && "oppo".equals(bid.getMedia())
+                && bid.getTagId() != null
+                && !bid.getTagId().isBlank();
+    }
+
     public void shrinkAfterClick(OppoBidRecord record) {
         try {
             String slimPayload = objectMapper.writeValueAsString(record.toPostClickRecord());

src/main/java/com/adx/tencent/storage/RedisHotStore.java

@@ -204,7 +204,14 @@ public class RedisHotStore {
             String payload = payloads != null ? payloads.get(i) : null;
             String qk = keyToQk.get(keys.get(i));
             if (payload != null) {
-                result.put(qk, parseBidRecord(payload));
+                BidRecord bid = parseBidRecord(payload);
+                if (isCompleteAttribution(bid)) {
+                    result.put(qk, bid);
+                } else if (qk != null) {
+                    missing.add(qk);
+                    log.warn("[RedisHotStore] incomplete Tencent cached bid, falling back to TiDB | qk={} | media={} | tagId={}",
+                            qk, bid == null ? null : bid.getMedia(), bid == null ? null : bid.getTagId());
+                }
             } else if (qk != null) {
                 missing.add(qk);
             }
@@ -219,6 +226,13 @@ public class RedisHotStore {
         return result;
     }
 
+    private static boolean isCompleteAttribution(BidRecord bid) {
+        return bid != null
+                && "tencent".equals(bid.getMedia())
+                && bid.getTagId() != null
+                && !bid.getTagId().isBlank();
+    }
+
     public BidRecord findBidByMediaTrace(String media, String traceId) {
         String payload = redis.opsForValue().get(mediaTraceKey(media, traceId));
         if (payload == null) return null;

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

@@ -192,7 +192,14 @@ public class VivoHotStore {
             String payload = payloads != null ? payloads.get(i) : null;
             String qk = keyToQk.get(keys.get(i));
             if (payload != null) {
-                result.put(qk, parseBidRecord(payload));
+                VivoBidRecord bid = parseBidRecord(payload);
+                if (isCompleteAttribution(bid)) {
+                    result.put(qk, bid);
+                } else if (qk != null) {
+                    missing.add(qk);
+                    log.warn("[VivoHotStore] incomplete cached bid, falling back to TiDB | qk={} | media={} | tagId={}",
+                            qk, bid == null ? null : bid.getMedia(), bid == null ? null : bid.getTagId());
+                }
             } else if (qk != null) {
                 missing.add(qk);
             }
@@ -205,6 +212,13 @@ public class VivoHotStore {
         return result;
     }
 
+    private static boolean isCompleteAttribution(VivoBidRecord bid) {
+        return bid != null
+                && "vivo".equals(bid.getMedia())
+                && bid.getTagId() != null
+                && !bid.getTagId().isBlank();
+    }
+
     public boolean markConversionSeen(String dedupeKey) {
         if (dedupeKey == null || dedupeKey.isBlank()) {
             return false;

src/test/java/com/adx/tencent/conversionsync/OtherMediaAttributionTest.java

@@ -0,0 +1,109 @@
+package com.adx.tencent.conversionsync;
+
+import com.adx.tencent.baidu.model.PaymentInfo;
+import com.adx.tencent.honor.model.HonorBidRecord;
+import com.adx.tencent.honor.store.HonorColdStore;
+import com.adx.tencent.honor.store.HonorHotStore;
+import com.adx.tencent.oppo.model.OppoBidRecord;
+import com.adx.tencent.oppo.store.OppoColdStore;
+import com.adx.tencent.oppo.store.OppoHotStore;
+import com.adx.tencent.storage.RedisHotStore;
+import com.adx.tencent.storage.TiDBColdStore;
+import com.adx.tencent.storage.model.BidRecord;
+import com.adx.tencent.storage.model.ConversionRecord;
+import com.adx.tencent.vivo.model.VivoBidRecord;
+import com.adx.tencent.vivo.store.VivoColdStore;
+import com.adx.tencent.vivo.store.VivoHotStore;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class OtherMediaAttributionTest {
+    private final ObjectMapper mapper = JsonMapper.builder()
+            .findAndAddModules()
+            .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
+            .build();
+
+    @Test
+    void incompleteTencentCacheFallsBackToColdRecord() throws Exception {
+        StringRedisTemplate redis = mock(StringRedisTemplate.class);
+        @SuppressWarnings("unchecked") ValueOperations<String, String> values = mock(ValueOperations.class);
+        TiDBColdStore cold = mock(TiDBColdStore.class);
+        BidRecord incomplete = new BidRecord(); incomplete.setQk("qk");
+        BidRecord complete = new BidRecord(); complete.setQk("qk"); complete.setMedia("tencent"); complete.setTagId("tag");
+        when(redis.opsForValue()).thenReturn(values);
+        when(values.multiGet(List.of("adx:tencent:bid:qk"))).thenReturn(List.of(mapper.writeValueAsString(incomplete)));
+        when(cold.getBidsByQks(List.of("qk"))).thenReturn(List.of(complete));
+
+        RedisHotStore store = new RedisHotStore(redis, mapper, cold, "adx:tencent:", "events", Duration.ofHours(24), 50_000);
+        assertSame(complete, store.findBidsByQks(List.of("qk")).get("qk"));
+    }
+
+    @Test
+    void incompleteHonorCacheFallsBackToColdRecord() throws Exception {
+        StringRedisTemplate redis = mock(StringRedisTemplate.class);
+        @SuppressWarnings("unchecked") ValueOperations<String, String> values = mock(ValueOperations.class);
+        HonorColdStore cold = mock(HonorColdStore.class);
+        HonorBidRecord incomplete = new HonorBidRecord(); incomplete.setQk("qk");
+        HonorBidRecord complete = new HonorBidRecord(); complete.setQk("qk"); complete.setMedia("honor"); complete.setTagId("tag");
+        when(redis.opsForValue()).thenReturn(values);
+        when(values.multiGet(List.of("adx:honor:bid:qk"))).thenReturn(List.of(mapper.writeValueAsString(incomplete)));
+        when(cold.getBidsByQks(List.of("qk"))).thenReturn(List.of(complete));
+
+        HonorHotStore store = new HonorHotStore(redis, mapper, cold, "adx:honor:", "events", Duration.ofHours(24), 50_000);
+        assertSame(complete, store.findBidsByQks(List.of("qk")).get("qk"));
+    }
+
+    @Test
+    void incompleteVivoCacheFallsBackToColdRecord() throws Exception {
+        StringRedisTemplate redis = mock(StringRedisTemplate.class);
+        @SuppressWarnings("unchecked") ValueOperations<String, String> values = mock(ValueOperations.class);
+        VivoColdStore cold = mock(VivoColdStore.class);
+        VivoBidRecord incomplete = new VivoBidRecord(); incomplete.setQk("qk");
+        VivoBidRecord complete = new VivoBidRecord(); complete.setQk("qk"); complete.setMedia("vivo"); complete.setTagId("tag");
+        when(redis.opsForValue()).thenReturn(values);
+        when(values.multiGet(List.of("adx:vivo:bid:qk"))).thenReturn(List.of(mapper.writeValueAsString(incomplete)));
+        when(cold.getBidsByQks(List.of("qk"))).thenReturn(List.of(complete));
+
+        VivoHotStore store = new VivoHotStore(redis, mapper, cold, "adx:vivo:", "events", Duration.ofHours(24), 50_000);
+        assertSame(complete, store.findBidsByQks(List.of("qk")).get("qk"));
+    }
+
+    @Test
+    void incompleteOppoCacheFallsBackToColdRecord() throws Exception {
+        StringRedisTemplate redis = mock(StringRedisTemplate.class);
+        @SuppressWarnings("unchecked") ValueOperations<String, String> values = mock(ValueOperations.class);
+        OppoColdStore cold = mock(OppoColdStore.class);
+        OppoBidRecord incomplete = new OppoBidRecord(); incomplete.setQk("qk");
+        OppoBidRecord complete = new OppoBidRecord(); complete.setQk("qk"); complete.setTagId("tag");
+        when(redis.opsForValue()).thenReturn(values);
+        when(values.multiGet(List.of("adx:oppo:bid:qk"))).thenReturn(List.of(mapper.writeValueAsString(incomplete)));
+        when(cold.getBidsByQks(List.of("qk"))).thenReturn(List.of(complete));
+
+        OppoHotStore store = new OppoHotStore(redis, mapper, cold, "events", Duration.ofHours(24), 50_000);
+        assertSame(complete, store.findBidsByQks(List.of("qk")).get("qk"));
+    }
+
+    @Test
+    void oppoConversionKeepsMediaAndTagId() {
+        PaymentInfo payment = new PaymentInfo();
+        payment.setQk("qk"); payment.setDate("20261016"); payment.setDeviceId("device"); payment.setAct(1);
+        OppoBidRecord bid = new OppoBidRecord(); bid.setTagId("jlt108");
+
+        ConversionRecord record = ConversionSyncService.toOppoConversionRecord(payment, bid);
+
+        assertEquals("oppo", record.getMedia());
+        assertEquals("jlt108", record.getTagId());
+    }
+}

src/test/java/com/adx/tencent/kuaishou/store/KuaishouHotStoreTest.java

@@ -0,0 +1,65 @@
+package com.adx.tencent.kuaishou.store;
+
+import com.adx.tencent.kuaishou.model.KuaishouBidRecord;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class KuaishouHotStoreTest {
+    private final ObjectMapper mapper = JsonMapper.builder().findAndAddModules().build();
+    private final StringRedisTemplate redis = mock(StringRedisTemplate.class);
+    @SuppressWarnings("unchecked")
+    private final ValueOperations<String, String> values = mock(ValueOperations.class);
+    private final KuaishouColdStore coldStore = mock(KuaishouColdStore.class);
+
+    @Test
+    void incompleteCachedBidFallsBackToColdRecordForConversionAttribution() throws Exception {
+        KuaishouBidRecord incomplete = new KuaishouBidRecord();
+        incomplete.setQk("qk");
+
+        KuaishouBidRecord complete = new KuaishouBidRecord();
+        complete.setQk("qk");
+        complete.setMedia("kuaishou");
+        complete.setTagId("jlt108");
+
+        when(redis.opsForValue()).thenReturn(values);
+        when(values.multiGet(List.of("adx:kuaishou:bid:qk")))
+                .thenReturn(List.of(mapper.writeValueAsString(incomplete)));
+        when(coldStore.getBidsByQks(List.of("qk"))).thenReturn(List.of(complete));
+
+        KuaishouHotStore store = new KuaishouHotStore(redis, mapper, coldStore,
+                "adx:kuaishou:", "adx:kuaishou:events", Duration.ofHours(24), 50_000);
+
+        assertSame(complete, store.findBidsByQks(List.of("qk")).get("qk"));
+        verify(coldStore).getBidsByQks(List.of("qk"));
+    }
+
+    @Test
+    void completeCachedBidDoesNotQueryColdStore() throws Exception {
+        KuaishouBidRecord complete = new KuaishouBidRecord();
+        complete.setQk("qk");
+        complete.setMedia("kuaishou");
+        complete.setTagId("jlt108");
+
+        when(redis.opsForValue()).thenReturn(values);
+        when(values.multiGet(List.of("adx:kuaishou:bid:qk")))
+                .thenReturn(List.of(mapper.writeValueAsString(complete)));
+
+        KuaishouHotStore store = new KuaishouHotStore(redis, mapper, coldStore,
+                "adx:kuaishou:", "adx:kuaishou:events", Duration.ofHours(24), 50_000);
+
+        assertEquals("jlt108", store.findBidsByQks(List.of("qk")).get("qk").getTagId());
+        verify(coldStore, org.mockito.Mockito.never()).getBidsByQks(org.mockito.ArgumentMatchers.anyList());
+    }
+}