yumeng 1 неделя назад
Родитель
Сommit
ef92ba5928

+ 82 - 1
src/main/java/com/adx/tencent/conversionsync/ConversionSyncService.java

@@ -883,8 +883,13 @@ public class ConversionSyncService {
         }
         }
 
 
         if (task.isHonor()) {
         if (task.isHonor()) {
+            int deductionRate = resolveHonorDeductionRate(task.honorBid, task.payment.getAct());
+            if (handleHonorDeduction(task)) {
+                return CallbackDispatchResult.honor(task, toCallbackRecord(task, buildDeductedHonorCallbackRecord(task, deductionRate)));
+            }
             HonorMediaCallbackRecord honorRecord = honorClient.sendConversion(
             HonorMediaCallbackRecord honorRecord = honorClient.sendConversion(
                     task.honorBid, task.payment, task.honorConversionId);
                     task.honorBid, task.payment, task.honorConversionId);
+            honorRecord.setDeductionRate(deductionRate);
             return CallbackDispatchResult.honor(task, toCallbackRecord(task, honorRecord));
             return CallbackDispatchResult.honor(task, toCallbackRecord(task, honorRecord));
         }
         }
 
 
@@ -1254,6 +1259,56 @@ public class ConversionSyncService {
         return record;
         return record;
     }
     }
 
 
+    private boolean handleHonorDeduction(CallbackTask task) {
+        if (!task.isHonor() || task.honorBid == null || honorHotStore == null) {
+            return false;
+        }
+        int deductionRate = resolveHonorDeductionRate(task.honorBid, task.payment.getAct());
+        if (deductionRate <= 0) {
+            return false;
+        }
+        if (!honorHotStore.markConversionSeen(task.callbackKey)) {
+            log.info("[ConversionSync] 荣耀转化已见过,跳过重复扣量判断 | callbackKey={} | qk={} | accountId={}",
+                    task.callbackKey, task.payment.getQk(), safe(task.honorBid.getAccountId()));
+            return false;
+        }
+        HonorHotStore.DeductionDecision decision = honorHotStore.evaluateDeduction(
+                task.honorBid.getAccountId(),
+                task.payment.getAct(),
+                task.payment.getDate(),
+                deductionRate
+        );
+        boolean deduct = decision.isShouldDeduct();
+        log.info("[ConversionSync] 荣耀转化扣量判定 | qk={} | act={} | accountId={} | tagId={} | rate={} | totalSeen={} | totalDeducted={} | callbackKey={} | deduct={}",
+                task.payment.getQk(), task.payment.getAct(), safe(task.honorBid.getAccountId()),
+                safe(task.honorBid.getTagId()), deductionRate,
+                decision.getTotalSeen(), decision.getTotalDeducted(), task.callbackKey, deduct);
+        return deduct;
+    }
+
+    private HonorMediaCallbackRecord buildDeductedHonorCallbackRecord(CallbackTask task, int deductionRate) {
+        HonorMediaCallbackRecord record = new HonorMediaCallbackRecord();
+        Instant now = Instant.now();
+        record.setDedupeKey(task.callbackKey);
+        record.setQk(task.payment.getQk());
+        record.setTagId(task.honorBid.getTagId());
+        record.setCallbackUrl("");
+        record.setEventType(task.payment.getAct());
+        record.setEventTimeMs(now.toEpochMilli());
+        record.setPurchase(Math.max(task.payment.getPayment(), 0D));
+        record.setDeductionRate(deductionRate);
+        record.setDispatchStatus(HonorMediaCallbackRecord.DISPATCH_STATUS_DEDUCTED);
+        record.setStatus(0);
+        record.setOk(false);
+        record.setAttempt(1);
+        record.setErrorMessage("deducted_by_rate");
+        record.setCreatedAt(now);
+        log.info("[ConversionSync] 荣耀转化扣量 | qk={} | act={} | rate={} | tagId={} | accountId={} | callbackKey={}",
+                task.payment.getQk(), task.payment.getAct(), deductionRate,
+                task.honorBid.getTagId(), safe(task.honorBid.getAccountId()), task.callbackKey);
+        return record;
+    }
+
     private boolean handleVivoDeduction(CallbackTask task) {
     private boolean handleVivoDeduction(CallbackTask task) {
         if (!task.isVivo() || task.vivoBid == null || vivoHotStore == null) {
         if (!task.isVivo() || task.vivoBid == null || vivoHotStore == null) {
             return false;
             return false;
@@ -1371,6 +1426,29 @@ public class ConversionSyncService {
         return 0;
         return 0;
     }
     }
 
 
+    private static int resolveHonorDeductionRate(HonorBidRecord bid, int act) {
+        if (bid == null || bid.getMediaParams() == null) {
+            return 0;
+        }
+        if (act == 2001) {
+            String rawActRate = firstParam(bid.getMediaParams(),
+                    HonorBidRecord.ACT_2001_DEDUCTION_RATE_KEY,
+                    "act2001DeductionRate",
+                    "deductionRate2001",
+                    "deduction_rate_2001");
+            Integer actRate = parseDeductionRate(rawActRate);
+            if (actRate != null) {
+                return actRate;
+            }
+        }
+        String rawRate = firstParam(bid.getMediaParams(),
+                HonorBidRecord.DEDUCTION_RATE_KEY,
+                "deductionRate",
+                "rate");
+        Integer rate = parseDeductionRate(rawRate);
+        return rate != null ? rate : 0;
+    }
+
     private int resolveVivoDeductionRate(VivoBidRecord bid) {
     private int resolveVivoDeductionRate(VivoBidRecord bid) {
         if (bid == null || bid.getMediaParams() == null) {
         if (bid == null || bid.getMediaParams() == null) {
             return 0;
             return 0;
@@ -1405,12 +1483,14 @@ public class ConversionSyncService {
         callback.setEventType(record.getEventType());
         callback.setEventType(record.getEventType());
         callback.setEventTimeMs(record.getEventTimeMs());
         callback.setEventTimeMs(record.getEventTimeMs());
         callback.setPurchase(record.getPurchase());
         callback.setPurchase(record.getPurchase());
+        callback.setDeductionRate(resolveHonorDeductionRate(task.honorBid, record.getEventType()));
         callback.setStatus(record.getStatus());
         callback.setStatus(record.getStatus());
         callback.setOk(record.isOk());
         callback.setOk(record.isOk());
         callback.setAttempt(record.getAttempt());
         callback.setAttempt(record.getAttempt());
         callback.setResponseBody(record.getResponseBody());
         callback.setResponseBody(record.getResponseBody());
         callback.setErrorMessage(record.getErrorMessage());
         callback.setErrorMessage(record.getErrorMessage());
         callback.setRequestBody(record.getRequestBody());
         callback.setRequestBody(record.getRequestBody());
+        callback.setDispatchStatus(record.getDispatchStatus());
         callback.setCreatedAt(record.getCreatedAt());
         callback.setCreatedAt(record.getCreatedAt());
         return callback;
         return callback;
     }
     }
@@ -1477,7 +1557,8 @@ public class ConversionSyncService {
         callbackRecord.setResponseBody(honorRecord.getResponseBody());
         callbackRecord.setResponseBody(honorRecord.getResponseBody());
         callbackRecord.setErrorMessage(honorRecord.getErrorMessage());
         callbackRecord.setErrorMessage(honorRecord.getErrorMessage());
         callbackRecord.setRequestBody(honorRecord.getRequestBody());
         callbackRecord.setRequestBody(honorRecord.getRequestBody());
-        callbackRecord.setTrackingVersion("v1");
+        callbackRecord.setDispatchStatus(honorRecord.getDispatchStatus());
+        callbackRecord.setTrackingVersion(String.valueOf(honorRecord.getDeductionRate()));
         callbackRecord.setCreatedAt(honorRecord.getCreatedAt());
         callbackRecord.setCreatedAt(honorRecord.getCreatedAt());
         return callbackRecord;
         return callbackRecord;
     }
     }

+ 3 - 0
src/main/java/com/adx/tencent/honor/client/HonorClient.java

@@ -42,6 +42,7 @@ public class HonorClient {
         record.setTagId(bid.getTagId());
         record.setTagId(bid.getTagId());
         record.setEventType(payment.getAct());
         record.setEventType(payment.getAct());
         record.setPurchase(payment.getPayment());
         record.setPurchase(payment.getPayment());
+        record.setDispatchStatus(HonorMediaCallbackRecord.DISPATCH_STATUS_SENT);
         record.setAttempt(1);
         record.setAttempt(1);
         record.setCreatedAt(Instant.now());
         record.setCreatedAt(Instant.now());
 
 
@@ -137,6 +138,8 @@ public class HonorClient {
         copy.setEventType(src.getEventType());
         copy.setEventType(src.getEventType());
         copy.setEventTimeMs(src.getEventTimeMs());
         copy.setEventTimeMs(src.getEventTimeMs());
         copy.setPurchase(src.getPurchase());
         copy.setPurchase(src.getPurchase());
+        copy.setDeductionRate(src.getDeductionRate());
+        copy.setDispatchStatus(src.getDispatchStatus());
         copy.setStatus(src.getStatus());
         copy.setStatus(src.getStatus());
         copy.setOk(src.isOk());
         copy.setOk(src.isOk());
         copy.setAttempt(src.getAttempt());
         copy.setAttempt(src.getAttempt());

+ 25 - 0
src/main/java/com/adx/tencent/honor/controller/HonorTrackingController.java

@@ -435,9 +435,34 @@ public class HonorTrackingController {
             String platform = HonorPlacement.normalizePlatform(os);
             String platform = HonorPlacement.normalizePlatform(os);
             if (platform != null) params.put("platform", platform);
             if (platform != null) params.put("platform", platform);
         }
         }
+        normalizeDeductionParams(params);
         return params;
         return params;
     }
     }
 
 
+    private void normalizeDeductionParams(Map<String, String> params) {
+        String rate = firstParam(params, HonorBidRecord.DEDUCTION_RATE_KEY, "deductionRate", "rate");
+        params.put(HonorBidRecord.DEDUCTION_RATE_KEY, String.valueOf(parseDeductionRate(rate)));
+        String act2001Rate = firstParam(params,
+                HonorBidRecord.ACT_2001_DEDUCTION_RATE_KEY,
+                "act2001DeductionRate",
+                "deductionRate2001",
+                "deduction_rate_2001");
+        if (act2001Rate != null) {
+            params.put(HonorBidRecord.ACT_2001_DEDUCTION_RATE_KEY, String.valueOf(parseDeductionRate(act2001Rate)));
+        }
+    }
+
+    private static int parseDeductionRate(String value) {
+        if (value == null || value.isBlank()) {
+            return 0;
+        }
+        try {
+            return Math.max(0, Math.min(100, Integer.parseInt(value.trim())));
+        } catch (NumberFormatException ignored) {
+            return 0;
+        }
+    }
+
     private String requireTagId(Map<String, String> params) {
     private String requireTagId(Map<String, String> params) {
         String tagId = params.get("tagId");
         String tagId = params.get("tagId");
         if (tagId == null || tagId.isBlank()) {
         if (tagId == null || tagId.isBlank()) {

+ 9 - 0
src/main/java/com/adx/tencent/honor/model/HonorBidRecord.java

@@ -8,6 +8,9 @@ import java.util.List;
 import java.util.Map;
 import java.util.Map;
 
 
 public class HonorBidRecord {
 public class HonorBidRecord {
+    public static final String DEDUCTION_RATE_KEY = "deduction_rate";
+    public static final String ACT_2001_DEDUCTION_RATE_KEY = "act_2001_deduction_rate";
+
     private String qk;
     private String qk;
     private String media;
     private String media;
     private String mediaTraceId;
     private String mediaTraceId;
@@ -130,6 +133,12 @@ public class HonorBidRecord {
                 "oaid", "oaidMd5", "oaid_md5", "hash_oaid",
                 "oaid", "oaidMd5", "oaid_md5", "hash_oaid",
                 "dpid", "dpidMd5", "dpid_md5", "hash_android_id",
                 "dpid", "dpidMd5", "dpid_md5", "hash_android_id",
                 "muid",
                 "muid",
+                DEDUCTION_RATE_KEY,
+                ACT_2001_DEDUCTION_RATE_KEY,
+                "deductionRate",
+                "act2001DeductionRate",
+                "deductionRate2001",
+                "deduction_rate_2001",
                 "rta_tu",
                 "rta_tu",
                 "rta_order_id",
                 "rta_order_id",
                 "rta_request_order_id",
                 "rta_request_order_id",

+ 9 - 0
src/main/java/com/adx/tencent/honor/model/HonorMediaCallbackRecord.java

@@ -3,6 +3,9 @@ package com.adx.tencent.honor.model;
 import java.time.Instant;
 import java.time.Instant;
 
 
 public class HonorMediaCallbackRecord {
 public class HonorMediaCallbackRecord {
+    public static final String DISPATCH_STATUS_SENT = "SENT";
+    public static final String DISPATCH_STATUS_DEDUCTED = "DEDUCTED";
+
     private String dedupeKey;
     private String dedupeKey;
     private String qk;
     private String qk;
     private String tagId;
     private String tagId;
@@ -10,6 +13,8 @@ public class HonorMediaCallbackRecord {
     private int eventType;
     private int eventType;
     private long eventTimeMs;
     private long eventTimeMs;
     private double purchase;
     private double purchase;
+    private int deductionRate;
+    private String dispatchStatus = DISPATCH_STATUS_SENT;
     private int status;
     private int status;
     private boolean ok;
     private boolean ok;
     private int attempt;
     private int attempt;
@@ -32,6 +37,10 @@ public class HonorMediaCallbackRecord {
     public void setEventTimeMs(long v) { this.eventTimeMs = v; }
     public void setEventTimeMs(long v) { this.eventTimeMs = v; }
     public double getPurchase() { return purchase; }
     public double getPurchase() { return purchase; }
     public void setPurchase(double v) { this.purchase = v; }
     public void setPurchase(double v) { this.purchase = v; }
+    public int getDeductionRate() { return deductionRate; }
+    public void setDeductionRate(int v) { this.deductionRate = v; }
+    public String getDispatchStatus() { return dispatchStatus; }
+    public void setDispatchStatus(String v) { this.dispatchStatus = v; }
     public int getStatus() { return status; }
     public int getStatus() { return status; }
     public void setStatus(int v) { this.status = v; }
     public void setStatus(int v) { this.status = v; }
     public boolean isOk() { return ok; }
     public boolean isOk() { return ok; }

+ 104 - 1
src/main/java/com/adx/tencent/honor/service/HonorConversionSyncService.java

@@ -113,7 +113,7 @@ public class HonorConversionSyncService {
 
 
         List<CompletableFuture<HonorMediaCallbackRecord>> futures = new ArrayList<>();
         List<CompletableFuture<HonorMediaCallbackRecord>> futures = new ArrayList<>();
         for (CallbackWork work : toSend) {
         for (CallbackWork work : toSend) {
-            futures.add(CompletableFuture.supplyAsync(() -> honorClient.sendConversion(work.bid, work.payment, work.conversionId), executor));
+            futures.add(CompletableFuture.supplyAsync(() -> sendOrDeduct(work), executor));
         }
         }
 
 
         List<HonorMediaCallbackRecord> callbackRecords = new ArrayList<>();
         List<HonorMediaCallbackRecord> callbackRecords = new ArrayList<>();
@@ -129,6 +129,109 @@ public class HonorConversionSyncService {
         }
         }
     }
     }
 
 
+    private HonorMediaCallbackRecord sendOrDeduct(CallbackWork work) {
+        int deductionRate = resolveDeductionRate(work.bid, work.payment.getAct());
+        if (handleDeduction(work, deductionRate)) {
+            return buildDeductedCallbackRecord(work, deductionRate);
+        }
+        HonorMediaCallbackRecord record = honorClient.sendConversion(work.bid, work.payment, work.conversionId);
+        record.setDeductionRate(deductionRate);
+        return record;
+    }
+
+    private boolean handleDeduction(CallbackWork work, int deductionRate) {
+        if (deductionRate <= 0) {
+            return false;
+        }
+        if (!hotStore.markConversionSeen(work.dedupeKey)) {
+            log.info("[HonorConversionSync] 转化已见过,跳过重复扣量判断 | callbackKey={} | qk={} | accountId={}",
+                    work.dedupeKey, work.payment.getQk(), safe(work.bid.getAccountId()));
+            return false;
+        }
+        HonorHotStore.DeductionDecision decision = hotStore.evaluateDeduction(
+                work.bid.getAccountId(),
+                work.payment.getAct(),
+                work.payment.getDate(),
+                deductionRate
+        );
+        boolean deduct = decision.isShouldDeduct();
+        log.info("[HonorConversionSync] 转化扣量判定 | qk={} | act={} | accountId={} | tagId={} | rate={} | totalSeen={} | totalDeducted={} | callbackKey={} | deduct={}",
+                work.payment.getQk(), work.payment.getAct(), safe(work.bid.getAccountId()),
+                safe(work.bid.getTagId()), deductionRate,
+                decision.getTotalSeen(), decision.getTotalDeducted(), work.dedupeKey, deduct);
+        return deduct;
+    }
+
+    private HonorMediaCallbackRecord buildDeductedCallbackRecord(CallbackWork work, int deductionRate) {
+        HonorMediaCallbackRecord record = new HonorMediaCallbackRecord();
+        Instant now = Instant.now();
+        record.setDedupeKey(work.dedupeKey);
+        record.setQk(work.payment.getQk());
+        record.setTagId(work.bid.getTagId());
+        record.setCallbackUrl("");
+        record.setEventType(work.payment.getAct());
+        record.setEventTimeMs(now.toEpochMilli());
+        record.setPurchase(Math.max(work.payment.getPayment(), 0D));
+        record.setDeductionRate(deductionRate);
+        record.setDispatchStatus(HonorMediaCallbackRecord.DISPATCH_STATUS_DEDUCTED);
+        record.setStatus(0);
+        record.setOk(false);
+        record.setAttempt(1);
+        record.setErrorMessage("deducted_by_rate");
+        record.setCreatedAt(now);
+        log.info("[HonorConversionSync] 转化扣量 | qk={} | act={} | rate={} | tagId={} | accountId={} | callbackKey={}",
+                work.payment.getQk(), work.payment.getAct(), deductionRate,
+                work.bid.getTagId(), safe(work.bid.getAccountId()), work.dedupeKey);
+        return record;
+    }
+
+    private static int resolveDeductionRate(HonorBidRecord bid, int act) {
+        if (bid == null || bid.getMediaParams() == null) {
+            return 0;
+        }
+        if (act == 2001) {
+            Integer actRate = parseDeductionRate(firstParam(bid.getMediaParams(),
+                    HonorBidRecord.ACT_2001_DEDUCTION_RATE_KEY,
+                    "act2001DeductionRate",
+                    "deductionRate2001",
+                    "deduction_rate_2001"));
+            if (actRate != null) {
+                return actRate;
+            }
+        }
+        Integer rate = parseDeductionRate(firstParam(bid.getMediaParams(),
+                HonorBidRecord.DEDUCTION_RATE_KEY,
+                "deductionRate",
+                "rate"));
+        return rate == null ? 0 : rate;
+    }
+
+    private static Integer parseDeductionRate(String rawRate) {
+        if (rawRate == null || rawRate.isBlank()) {
+            return null;
+        }
+        try {
+            int parsed = Integer.parseInt(rawRate.trim());
+            return Math.max(0, Math.min(parsed, 100));
+        } catch (NumberFormatException ignored) {
+            return null;
+        }
+    }
+
+    private static String firstParam(Map<String, String> params, String... keys) {
+        for (String key : keys) {
+            String value = params.get(key);
+            if (value != null && !value.isBlank()) {
+                return value.trim();
+            }
+        }
+        return null;
+    }
+
+    private static String safe(String value) {
+        return value == null ? "" : value;
+    }
+
     private HonorConversionRecord toConversionRecord(PaymentInfo payment, HonorBidRecord bid) {
     private HonorConversionRecord toConversionRecord(PaymentInfo payment, HonorBidRecord bid) {
         HonorConversionRecord record = new HonorConversionRecord();
         HonorConversionRecord record = new HonorConversionRecord();
         record.setDedupeKey(conversionDedupeKey(payment));
         record.setDedupeKey(conversionDedupeKey(payment));

+ 2 - 0
src/main/java/com/adx/tencent/honor/service/HonorRetryService.java

@@ -54,6 +54,8 @@ public class HonorRetryService {
         copy.setEventType(src.getEventType());
         copy.setEventType(src.getEventType());
         copy.setEventTimeMs(src.getEventTimeMs());
         copy.setEventTimeMs(src.getEventTimeMs());
         copy.setPurchase(src.getPurchase());
         copy.setPurchase(src.getPurchase());
+        copy.setDeductionRate(src.getDeductionRate());
+        copy.setDispatchStatus(src.getDispatchStatus());
         copy.setStatus(src.getStatus());
         copy.setStatus(src.getStatus());
         copy.setOk(src.isOk());
         copy.setOk(src.isOk());
         copy.setAttempt(src.getAttempt());
         copy.setAttempt(src.getAttempt());

+ 10 - 0
src/main/java/com/adx/tencent/honor/store/HonorColdStore.java

@@ -85,6 +85,8 @@ public class HonorColdStore {
                     event_type INT NOT NULL,
                     event_type INT NOT NULL,
                     event_time_ms BIGINT NOT NULL,
                     event_time_ms BIGINT NOT NULL,
                     purchase DOUBLE,
                     purchase DOUBLE,
+                    deduction_rate INT NOT NULL DEFAULT 0,
+                    dispatch_status VARCHAR(32) NOT NULL DEFAULT 'SENT',
                     status INT NOT NULL,
                     status INT NOT NULL,
                     ok TINYINT(1) NOT NULL,
                     ok TINYINT(1) NOT NULL,
                     attempt INT NOT NULL DEFAULT 1,
                     attempt INT NOT NULL DEFAULT 1,
@@ -121,6 +123,14 @@ public class HonorColdStore {
                 stmt.execute("ALTER TABLE honor_tracking_reports ADD KEY idx_honor_tracking_reports_created_at_id (created_at, id)");
                 stmt.execute("ALTER TABLE honor_tracking_reports ADD KEY idx_honor_tracking_reports_created_at_id (created_at, id)");
             } catch (Exception ignored) {
             } catch (Exception ignored) {
             }
             }
+            try {
+                stmt.execute("ALTER TABLE honor_media_callbacks ADD COLUMN deduction_rate INT NOT NULL DEFAULT 0 COMMENT '扣量比例(0-100)' AFTER purchase");
+            } catch (Exception ignored) {
+            }
+            try {
+                stmt.execute("ALTER TABLE honor_media_callbacks ADD COLUMN dispatch_status VARCHAR(32) NOT NULL DEFAULT 'SENT' COMMENT '回传处理状态(SENT/DEDUCTED)' AFTER deduction_rate");
+            } catch (Exception ignored) {
+            }
             stmt.close();
             stmt.close();
         } catch (Exception e) {
         } catch (Exception e) {
             throw new RuntimeException("honor migrate failed", e);
             throw new RuntimeException("honor migrate failed", e);

+ 92 - 0
src/main/java/com/adx/tencent/honor/store/HonorHotStore.java

@@ -13,6 +13,7 @@ import org.springframework.data.redis.connection.stream.RecordId;
 import org.springframework.data.redis.connection.stream.StreamOffset;
 import org.springframework.data.redis.connection.stream.StreamOffset;
 import org.springframework.data.redis.connection.stream.StreamReadOptions;
 import org.springframework.data.redis.connection.stream.StreamReadOptions;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.script.DefaultRedisScript;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.LoggerFactory;
 
 
@@ -33,6 +34,8 @@ public class HonorHotStore {
 
 
     private static final Logger log = LoggerFactory.getLogger(HonorHotStore.class);
     private static final Logger log = LoggerFactory.getLogger(HonorHotStore.class);
     private static final Duration IMPRESSION_BID_TTL = Duration.ofHours(3);
     private static final Duration IMPRESSION_BID_TTL = Duration.ofHours(3);
+    private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
+    private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
 
     private final StringRedisTemplate redis;
     private final StringRedisTemplate redis;
     private final ObjectMapper objectMapper;
     private final ObjectMapper objectMapper;
@@ -41,6 +44,7 @@ public class HonorHotStore {
     private final String stream;
     private final String stream;
     private final Duration bidTtl;
     private final Duration bidTtl;
     private final long streamMaxLen;
     private final long streamMaxLen;
+    private final DefaultRedisScript<List> deductionDecisionScript;
     private final Set<String> initializedGroups = ConcurrentHashMap.newKeySet();
     private final Set<String> initializedGroups = ConcurrentHashMap.newKeySet();
 
 
     public HonorHotStore(StringRedisTemplate redis, ObjectMapper objectMapper,
     public HonorHotStore(StringRedisTemplate redis, ObjectMapper objectMapper,
@@ -53,6 +57,7 @@ public class HonorHotStore {
         this.stream = stream;
         this.stream = stream;
         this.bidTtl = bidTtl;
         this.bidTtl = bidTtl;
         this.streamMaxLen = streamMaxLen;
         this.streamMaxLen = streamMaxLen;
+        this.deductionDecisionScript = buildDeductionDecisionScript();
     }
     }
 
 
     public void recordBid(HonorBidRecord record) {
     public void recordBid(HonorBidRecord record) {
@@ -168,6 +173,35 @@ public class HonorHotStore {
         return result;
         return result;
     }
     }
 
 
+    public boolean markConversionSeen(String dedupeKey) {
+        if (dedupeKey == null || dedupeKey.isBlank()) {
+            return false;
+        }
+        Boolean created = redis.opsForValue().setIfAbsent(
+                conversionSeenKey(dedupeKey),
+                "1",
+                DEFAULT_CONVERSION_SEEN_TTL
+        );
+        return Boolean.TRUE.equals(created);
+    }
+
+    public DeductionDecision evaluateDeduction(String accountId, int act, String date, int deductionRate) {
+        String key = deductionCounterKey(accountId, act, date);
+        List result = redis.execute(
+                deductionDecisionScript,
+                Collections.singletonList(key),
+                String.valueOf(Math.max(deductionRate, 0)),
+                String.valueOf(DEFAULT_DEDUCTION_COUNTER_TTL.getSeconds())
+        );
+        if (result == null || result.size() < 3) {
+            throw new RuntimeException("evaluate honor deduction failed");
+        }
+        long total = toLong(result.get(0));
+        long deducted = toLong(result.get(1));
+        boolean shouldDeduct = toLong(result.get(2)) == 1L;
+        return new DeductionDecision(total, deducted, shouldDeduct);
+    }
+
     public void ensureGroup(String group) {
     public void ensureGroup(String group) {
         if (group == null || group.isBlank()) return;
         if (group == null || group.isBlank()) return;
         if (initializedGroups.contains(group)) return;
         if (initializedGroups.contains(group)) return;
@@ -265,6 +299,48 @@ public class HonorHotStore {
 
 
     private String bidKey(String qk) { return prefix + "bid:" + qk; }
     private String bidKey(String qk) { return prefix + "bid:" + qk; }
     private String mediaTraceKey(String traceId) { return prefix + "media:honor:" + traceId; }
     private String mediaTraceKey(String traceId) { return prefix + "media:honor:" + traceId; }
+    private String conversionSeenKey(String dedupeKey) { return prefix + "conv:seen:honor:" + dedupeKey; }
+
+    private String deductionCounterKey(String accountId, int act, String date) {
+        return String.format("%sdeduct:state:honor:acct:%s:act:%d:date:%s",
+                prefix,
+                sanitizePart(accountId, "unknown"),
+                act,
+                sanitizePart(date, "unknown"));
+    }
+
+    private static String sanitizePart(String value, String fallback) {
+        if (value == null || value.isBlank()) return fallback;
+        return value.replace(':', '_').trim();
+    }
+
+    private static long toLong(Object value) {
+        if (value instanceof Number number) return number.longValue();
+        return Long.parseLong(String.valueOf(value));
+    }
+
+    private DefaultRedisScript<List> buildDeductionDecisionScript() {
+        DefaultRedisScript<List> script = new DefaultRedisScript<>();
+        script.setResultType(List.class);
+        script.setScriptText("""
+                local key = KEYS[1]
+                local rate = tonumber(ARGV[1]) or 0
+                local ttl = tonumber(ARGV[2]) or 0
+                local total = redis.call('HINCRBY', key, 'total', 1)
+                local deducted = tonumber(redis.call('HGET', key, 'deducted') or '0')
+                local target = math.floor((total * rate) / 100)
+                local should_deduct = 0
+                if deducted < target then
+                    deducted = redis.call('HINCRBY', key, 'deducted', 1)
+                    should_deduct = 1
+                end
+                if ttl > 0 then
+                    redis.call('EXPIRE', key, ttl)
+                end
+                return { total, deducted, should_deduct }
+                """);
+        return script;
+    }
 
 
     private HonorBidRecord parseBidRecord(String payload) {
     private HonorBidRecord parseBidRecord(String payload) {
         try {
         try {
@@ -315,4 +391,20 @@ public class HonorHotStore {
         }
         }
         return msg == null ? "" : msg;
         return msg == null ? "" : msg;
     }
     }
+
+    public static class DeductionDecision {
+        private final long totalSeen;
+        private final long totalDeducted;
+        private final boolean shouldDeduct;
+
+        public DeductionDecision(long totalSeen, long totalDeducted, boolean shouldDeduct) {
+            this.totalSeen = totalSeen;
+            this.totalDeducted = totalDeducted;
+            this.shouldDeduct = shouldDeduct;
+        }
+
+        public long getTotalSeen() { return totalSeen; }
+        public long getTotalDeducted() { return totalDeducted; }
+        public boolean isShouldDeduct() { return shouldDeduct; }
+    }
 }
 }

+ 8 - 5
src/main/resources/mapper/HonorMediaCallbackMapper.xml

@@ -10,6 +10,8 @@
         <result property="eventType" column="event_type"/>
         <result property="eventType" column="event_type"/>
         <result property="eventTimeMs" column="event_time_ms"/>
         <result property="eventTimeMs" column="event_time_ms"/>
         <result property="purchase" column="purchase"/>
         <result property="purchase" column="purchase"/>
+        <result property="deductionRate" column="deduction_rate"/>
+        <result property="dispatchStatus" column="dispatch_status"/>
         <result property="status" column="status"/>
         <result property="status" column="status"/>
         <result property="ok" column="ok_val"/>
         <result property="ok" column="ok_val"/>
         <result property="attempt" column="attempt"/>
         <result property="attempt" column="attempt"/>
@@ -22,12 +24,12 @@
     <insert id="batchInsert">
     <insert id="batchInsert">
         INSERT INTO honor_media_callbacks (
         INSERT INTO honor_media_callbacks (
             dedupe_key, qk, tag_id, callback_url, event_type, event_time_ms, purchase,
             dedupe_key, qk, tag_id, callback_url, event_type, event_time_ms, purchase,
-            status, ok, attempt, response_body, error_message, request_body, created_at
+            deduction_rate, dispatch_status, status, ok, attempt, response_body, error_message, request_body, created_at
         ) VALUES
         ) VALUES
         <foreach collection="records" item="r" separator=",">
         <foreach collection="records" item="r" separator=",">
             (
             (
                 #{r.dedupeKey}, #{r.qk}, #{r.tagId}, #{r.callbackUrl}, #{r.eventType},
                 #{r.dedupeKey}, #{r.qk}, #{r.tagId}, #{r.callbackUrl}, #{r.eventType},
-                #{r.eventTimeMs}, #{r.purchase}, #{r.status}, #{r.ok}, #{r.attempt},
+                #{r.eventTimeMs}, #{r.purchase}, #{r.deductionRate}, #{r.dispatchStatus}, #{r.status}, #{r.ok}, #{r.attempt},
                 #{r.responseBody}, #{r.errorMessage}, #{r.requestBody}, #{r.createdAtTs}
                 #{r.responseBody}, #{r.errorMessage}, #{r.requestBody}, #{r.createdAtTs}
             )
             )
         </foreach>
         </foreach>
@@ -36,7 +38,7 @@
     <select id="selectSuccessfulDedupeKeys" resultType="string">
     <select id="selectSuccessfulDedupeKeys" resultType="string">
         SELECT DISTINCT dedupe_key
         SELECT DISTINCT dedupe_key
         FROM honor_media_callbacks
         FROM honor_media_callbacks
-        WHERE ok = 1
+        WHERE (ok = 1 OR dispatch_status = 'DEDUCTED')
           AND dedupe_key IN
           AND dedupe_key IN
           <foreach collection="dedupeKeys" item="dedupeKey" open="(" separator="," close=")">
           <foreach collection="dedupeKeys" item="dedupeKey" open="(" separator="," close=")">
               #{dedupeKey}
               #{dedupeKey}
@@ -45,7 +47,7 @@
 
 
     <select id="selectPendingCallbacks" resultMap="honorMediaCallbackResultMap">
     <select id="selectPendingCallbacks" resultMap="honorMediaCallbackResultMap">
         SELECT m.dedupe_key, m.qk, m.tag_id, m.callback_url, m.event_type, m.event_time_ms,
         SELECT m.dedupe_key, m.qk, m.tag_id, m.callback_url, m.event_type, m.event_time_ms,
-               m.purchase, m.status, m.ok AS ok_val, m.attempt, m.response_body, m.error_message,
+               m.purchase, m.deduction_rate, m.dispatch_status, m.status, m.ok AS ok_val, m.attempt, m.response_body, m.error_message,
                m.request_body, m.created_at
                m.request_body, m.created_at
         FROM honor_media_callbacks m
         FROM honor_media_callbacks m
         JOIN (
         JOIN (
@@ -55,9 +57,10 @@
             GROUP BY dedupe_key
             GROUP BY dedupe_key
         ) latest ON latest.id = m.id
         ) latest ON latest.id = m.id
         WHERE m.ok = 0
         WHERE m.ok = 0
+          AND COALESCE(m.dispatch_status, 'SENT') != 'DEDUCTED'
           AND NOT EXISTS (
           AND NOT EXISTS (
             SELECT 1 FROM honor_media_callbacks ok_cb
             SELECT 1 FROM honor_media_callbacks ok_cb
-            WHERE ok_cb.dedupe_key = m.dedupe_key AND ok_cb.ok = 1
+            WHERE ok_cb.dedupe_key = m.dedupe_key AND (ok_cb.ok = 1 OR ok_cb.dispatch_status = 'DEDUCTED')
           )
           )
         ORDER BY m.created_at ASC
         ORDER BY m.created_at ASC
         LIMIT #{limit}
         LIMIT #{limit}

+ 2 - 0
src/main/resources/schema-honor.sql

@@ -58,6 +58,8 @@ CREATE TABLE IF NOT EXISTS honor_media_callbacks (
     event_type INT NOT NULL COMMENT '事件类型(百度act)',
     event_type INT NOT NULL COMMENT '事件类型(百度act)',
     event_time_ms BIGINT NOT NULL COMMENT '事件时间(毫秒时间戳)',
     event_time_ms BIGINT NOT NULL COMMENT '事件时间(毫秒时间戳)',
     purchase DOUBLE COMMENT '付费金额',
     purchase DOUBLE COMMENT '付费金额',
+    deduction_rate INT NOT NULL DEFAULT 0 COMMENT '扣量比例(0-100)',
+    dispatch_status VARCHAR(32) NOT NULL DEFAULT 'SENT' COMMENT '回传处理状态(SENT/DEDUCTED)',
     status INT NOT NULL COMMENT 'HTTP响应状态码',
     status INT NOT NULL COMMENT 'HTTP响应状态码',
     ok TINYINT(1) NOT NULL COMMENT '是否成功(1=成功,0=失败)',
     ok TINYINT(1) NOT NULL COMMENT '是否成功(1=成功,0=失败)',
     attempt INT NOT NULL DEFAULT 1 COMMENT '尝试次数',
     attempt INT NOT NULL DEFAULT 1 COMMENT '尝试次数',