fix(selfop): discover warehouses with bounded validation and optimize queries

This commit is contained in:
2026-10-12 22:22:08 +08:00
parent e42d358e71
commit b71380de31
20 changed files with 419 additions and 53 deletions
@@ -1,6 +1,10 @@
package com.baiyue.lastleg.selfoperated.analytics.model;
import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.require;
import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.read;
import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.identity;
import java.util.Map;
import java.util.List;
/** Purpose is persisted on every task/run. Unknown purpose is never treated as business. */
public final class SelfopCollectionPolicy {
@@ -20,4 +24,27 @@ public final class SelfopCollectionPolicy {
public static String configuredPurpose(SelfopAnalyticsProperties config) {
String purpose=config.getCollectionMode();requireExecution(config,purpose);return purpose;
}
/** The action and warehouse are immutable task inputs, never taken from later configuration. */
public static boolean warehouseDiscovery(SelfopSyncTask task) {
require(task!=null&&"VALIDATION".equals(task.getExecutionPurpose())&&"VALIDATION".equals(task.getTaskType())
&&"ACCOUNT_DISCOVERY".equals(task.getTaskRole()),"SELFOP_VALIDATION_ENTRY_REQUIRED");
Map<?,?> query=read(task.getQueryScopeJson(),Map.class);
require(query!=null&&("WAREHOUSE_DISCOVERY".equals(query.get("action"))||"VALIDATION".equals(query.get("action"))),
"SELFOP_VALIDATION_ENTRY_REQUIRED");
if("WAREHOUSE_DISCOVERY".equals(query.get("action"))) {
require(!query.containsKey("warehouses")&&!query.containsKey("startTime")&&!query.containsKey("endTime"),
"SELFOP_WAREHOUSE_DISCOVERY_PARAMETERS");
return true;
}
return false;
}
public static String validationWarehouse(SelfopSyncTask task) {
require(!warehouseDiscovery(task),"SELFOP_VALIDATION_WAREHOUSE_REQUIRED");
Map<?,?> query=read(task.getQueryScopeJson(),Map.class);
require(query.get("warehouses") instanceof List<?> warehouses&&warehouses.size()==1
&&warehouses.get(0) instanceof String,"SELFOP_VALIDATION_WAREHOUSE_REQUIRED");
String warehouse=(String)((List<?>)query.get("warehouses")).get(0);
identity(warehouse,64);
return warehouse;
}
}
@@ -30,15 +30,12 @@ public final class AnalyticsInventorySql {
+" sales AS (SELECT customer_code,sku_code,SUM(outbound_quantity) observed_quantity,SUM(outbound_order_count) order_count,"
+"MIN(CASE WHEN quantity_unit='PCS' AND outbound_quantity>=0 THEN 1 ELSE 0 END) unit_valid FROM ll_selfop_sku_daily_sales "
+"WHERE sales_run_id=#{p.salesRun} AND"+scope(p)+" AND business_date>=#{p.from} AND business_date<#{p.until} GROUP BY customer_code,sku_code),"
+" last_sales AS (SELECT customer_code,sku_code,MAX(business_date) last_outbound FROM ll_selfop_sku_daily_sales WHERE sales_run_id=#{p.salesRun} AND"+scope(p)
+" AND business_date<=#{p.anchor} AND outbound_quantity>0 AND quantity_unit='PCS' GROUP BY customer_code,sku_code),"
+" values_by_sku AS (SELECT c.customer_code,c.sku_code,c.product_name,c.unit_valid,"
+"CASE WHEN c.sales_evidence IS NULL THEN c.stock_evidence WHEN c.stock_evidence IS NULL THEN c.sales_evidence ELSE LEAST(c.sales_evidence,c.stock_evidence) END evidence_at_utc,"
+stockValue("physical_quantity")+","+stockValue("available_quantity")+","+stockValue("frozen_quantity")+","+stockValue("transit_quantity")+","+stockValue("defective_physical_quantity")+",st.observed_at_utc,"
+"CASE WHEN c.sales_unit_valid=1 AND COALESCE(s.unit_valid,1)=1 AND #{p.complete30}=1 THEN COALESCE(s.observed_quantity,0) END owner_outbound,"
+"CASE WHEN c.sales_unit_valid=1 AND COALESCE(s.unit_valid,1)=1 AND #{p.salesKnown}=1 THEN s.observed_quantity END observed_outbound,s.order_count,l.last_outbound "
+"FROM profiles c LEFT JOIN stock st ON st.customer_code=c.customer_code AND st.sku_code=c.sku_code LEFT JOIN sales s ON s.customer_code=c.customer_code AND s.sku_code=c.sku_code "
+"LEFT JOIN last_sales l ON l.customer_code=c.customer_code AND l.sku_code=c.sku_code),"
+"CASE WHEN c.sales_unit_valid=1 AND COALESCE(s.unit_valid,1)=1 AND #{p.salesKnown}=1 THEN s.observed_quantity END observed_outbound,s.order_count "
+"FROM profiles c LEFT JOIN stock st ON st.customer_code=c.customer_code AND st.sku_code=c.sku_code LEFT JOIN sales s ON s.customer_code=c.customer_code AND s.sku_code=c.sku_code),"
+" qualified AS (SELECT v.*,CASE WHEN evidence_at_utc<=CONVERT_TZ(CAST(#{p.from} AS DATETIME),'+08:00','+00:00') THEN 1 ELSE 0 END observation_eligible,"
+"CASE WHEN unit_valid=1 AND evidence_at_utc<=CONVERT_TZ(CAST(#{p.from} AS DATETIME),'+08:00','+00:00') THEN owner_outbound END outbound_quantity30d FROM values_by_sku v),"
+" calculated AS (SELECT q.*,CASE WHEN available_quantity=0 THEN 0 WHEN available_quantity>0 AND outbound_quantity30d>0 THEN available_quantity*30/outbound_quantity30d END coverage_days,"
@@ -59,7 +56,17 @@ public final class AnalyticsInventorySql {
case "PHYSICAL_DESC" -> "physical_quantity";case "COVERAGE_DAYS_DESC" -> "coverage_days";case "OUTBOUND_30D_ASC" -> "outbound_quantity30d";default -> "available_quantity";};
return field+" IS NULL,"+field+("OUTBOUND_30D_ASC".equals(o.sortValue())?" ASC":" DESC")+",sku_code ASC,customer_code ASC";}
public static String inventoryPage(Map<String,Object> args){var p=(AnalyticsSqlRequest)args.get("p");var o=(AnalyticsInventoryOptions)args.get("o");
return typed(cte(p)+"SELECT customer_code,sku_code,product_name,unit_valid,evidence_at_utc,physical_quantity,available_quantity,frozen_quantity,transit_quantity,defective_physical_quantity,observed_at_utc,owner_outbound,observed_outbound,order_count,last_outbound,observation_eligible,outbound_quantity30d,coverage_days,no_outbound,above_threshold,unassessable FROM result"+filter(o)+" ORDER BY "+order(o)+" LIMIT #{p.size} OFFSET #{p.offset}");}
// The full candidate set is filtered and sorted first. Historical dates enrich only
// those immutable keys selected for this page and never participate in qualification.
String paged=cte(p)+", paged AS (SELECT * FROM result"+filter(o)+" ORDER BY "+order(o)+" LIMIT #{p.size} OFFSET #{p.offset}),"
+" last_sales AS (SELECT d.customer_code,d.sku_code,MAX(d.business_date) last_outbound FROM ll_selfop_sku_daily_sales d "
+"JOIN paged k ON k.customer_code=d.customer_code AND k.sku_code=d.sku_code WHERE d.sales_run_id=#{p.salesRun} "
+"AND d.source_account=#{p.account} AND d.warehouse_code=#{p.warehouse} AND d.business_date<=#{p.anchor} "
+"AND d.outbound_quantity>0 AND d.quantity_unit='PCS' GROUP BY d.customer_code,d.sku_code) ";
String outerOrder=order(o).replace("physical_quantity","k.physical_quantity").replace("coverage_days","k.coverage_days")
.replace("outbound_quantity30d","k.outbound_quantity30d").replace("available_quantity","k.available_quantity")
.replace("sku_code","k.sku_code").replace("customer_code","k.customer_code");
return typed(paged+"SELECT k.customer_code,k.sku_code,k.product_name,k.unit_valid,k.evidence_at_utc,k.physical_quantity,k.available_quantity,k.frozen_quantity,k.transit_quantity,k.defective_physical_quantity,k.observed_at_utc,k.owner_outbound,k.observed_outbound,k.order_count,l.last_outbound,k.observation_eligible,k.outbound_quantity30d,k.coverage_days,k.no_outbound,k.above_threshold,k.unassessable FROM paged k LEFT JOIN last_sales l ON l.customer_code=k.customer_code AND l.sku_code=k.sku_code ORDER BY "+outerOrder);}
public static String inventoryCount(Map<String,Object> args){var p=(AnalyticsSqlRequest)args.get("p");var o=(AnalyticsInventoryOptions)args.get("o");
return typed("ALL".equals(o.attentionValue())?"SELECT COUNT(*) FROM ("+keys(p)+") k":cte(p)+"SELECT COUNT(*) FROM result"+filter(o));}
public static String ownerTotals(Map<String,Object> args){var p=(AnalyticsSqlRequest)args.get("p");StringBuilder sql=new StringBuilder(cte(p)).append("SELECT COUNT(*) sku_count");
@@ -132,10 +132,11 @@ public class AnalyticsQueryService {
}
private AnalyticsResponses.Page<AnalyticsResponses.Metric> metrics(AnalyticsQuery q,String kind,boolean hot){
Snapshot s=context(q,false,false,hot); AnalyticsSqlRequest p=request(s,kind);
long total=mapper.count(p);
List<AnalyticsReadMapper.RankingStat> statistics=mapper.rankingStats(p);
long total=statistics.isEmpty()?0:statistics.get(0).total();
List<AnalyticsResponses.Metric> rows=q.offset()>=total?List.of():mapper.page(p).stream().map(r->metric(s,r)).toList();
Map<String,Long> excluded=new TreeMap<>();
if(!"INVENTORY".equals(kind))for(var e:mapper.excluded(p))excluded.put(e.reason(),e.total());
for(var e:statistics)if(e.reason()!=null)excluded.put(e.reason(),e.excludedCount());
return page(s,rows,total,excluded);
}
private AnalyticsResponses.SalesSummary summary(Snapshot s,Aggregate r){
@@ -20,9 +20,9 @@ public interface AnalyticsReadMapper {
List<AnalyticsReadRows.Aggregate> page(@Param("p") AnalyticsSqlRequest request);
@SelectProvider(type=AnalyticsReadSql.class,method="count")
long count(@Param("p") AnalyticsSqlRequest request);
record Exclusion(String reason,Long total){}
@SelectProvider(type=AnalyticsReadSql.class,method="excluded")
List<Exclusion> excluded(@Param("p") AnalyticsSqlRequest request);
record RankingStat(Long total,String reason,Long excludedCount){}
@SelectProvider(type=AnalyticsReadSql.class,method="rankingStats")
List<RankingStat> rankingStats(@Param("p") AnalyticsSqlRequest request);
@Select("SELECT customer_code,sku_code,product_name,quantity_unit,first_seen_at_utc,earliest_evidence_at_utc,"
+"evidence_type,evidence_run_id,last_seen_run_id FROM ll_selfop_sku_profile WHERE profile_run_id=#{run} "
+"AND source_account=#{account} AND warehouse_code=#{warehouse} AND customer_code=#{customer} AND sku_code=#{sku}")
@@ -29,6 +29,7 @@ public final class AnalyticsReadSql {
return sql;
}
private static String cte(String kind){
if("HOT".equals(kind)||"SLOW".equals(kind))return rankingCte();
String localEvidence="CONVERT_TZ(evidence_at_utc,'+00:00','+08:00')";
String observedStart="CONVERT_TZ(GREATEST(evidence_at_utc,#{p.coveredFrom}),'+00:00','+08:00')";
String observedEnd="CONVERT_TZ(#{p.coveredUntil},'+00:00','+08:00')";
@@ -84,9 +85,14 @@ public final class AnalyticsReadSql {
+"AND st.warehouse_code=#{p.warehouse} AND st.customer_code=c.customer_code AND st.sku_code=c.sku_code AND st.product_quality=0) ";
if("SUMMARY".equals(kind)){
int start=sql.indexOf(" metric_sales AS (");int end=sql.indexOf(" counted AS (",start);
sql=sql.substring(0,start)+" metric_sales AS (SELECT customer_code,sku_code,CAST(NULL AS DECIMAL(20,6)) window_quantity,"
+"CAST(NULL AS DECIMAL(20,0)) window_orders,CAST(NULL AS DECIMAL(20,6)) sales30_quantity,"
+"CAST(NULL AS DECIMAL(20,6)) today_quantity,CAST(NULL AS CHAR(32)) min_unit,CAST(NULL AS CHAR(32)) max_unit FROM bounds WHERE 1=0),"+sql.substring(end);
// An empty aggregate joined back through bounds can still be merged into
// another candidate cross product by MySQL. Summary has no metric join at all.
sql=sql.substring(0,start)+sql.substring(end);
sql=sql.replace("AND (m.min_unit IS NULL OR (m.min_unit=b.quantity_unit AND m.max_unit=b.quantity_unit)) ","")
.replace("m.window_quantity,m.window_orders,m.sales30_quantity,m.today_quantity",
"CAST(NULL AS DECIMAL(20,6)) window_quantity,CAST(NULL AS SIGNED) window_orders,"
+"CAST(NULL AS DECIMAL(20,6)) sales30_quantity,CAST(NULL AS DECIMAL(20,6)) today_quantity")
.replace("LEFT JOIN metric_sales m ON m.customer_code=b.customer_code AND m.sku_code=b.sku_code","");
// A sales chart does not require inventory quantities or freshness. Keep only profile evidence.
int stockJoin=sql.indexOf("LEFT JOIN ll_selfop_inventory_snapshot st");
sql=sql.substring(0,stockJoin)+"LEFT JOIN (SELECT CAST(NULL AS DECIMAL(20,0)) snapshot_run_id,CAST(NULL AS CHAR(64)) source_account,"
@@ -97,6 +103,32 @@ public final class AnalyticsReadSql {
}
return sql;
}
/** Ranking consumes the 30-day aggregate, not chart day-span counts. Its 7/30-day
* window is contained in this scan, so the same unit proof covers both windows. */
private static String rankingCte(){
String valid="c.profile_unit_valid=1 AND (m.min_unit IS NULL OR (m.min_unit=c.quantity_unit AND m.max_unit=c.quantity_unit))";
return candidates(false)+" metric_sales AS (SELECT customer_code,sku_code,"
+"SUM(CASE WHEN business_date>=#{p.windowFrom} AND business_date<#{p.anchor} THEN outbound_quantity ELSE 0 END) window_quantity,"
+"SUM(CASE WHEN business_date>=#{p.windowFrom} AND business_date<#{p.anchor} THEN outbound_order_count ELSE 0 END) window_orders,"
+"SUM(CASE WHEN business_date<#{p.anchor} THEN outbound_quantity ELSE 0 END) sales30_quantity,"
+"SUM(CASE WHEN business_date=#{p.anchor} THEN outbound_quantity ELSE NULL END) today_quantity,MIN(quantity_unit) min_unit,MAX(quantity_unit) max_unit "
+"FROM ll_selfop_sku_daily_sales WHERE sales_run_id=#{p.salesRun} AND"+KEY_FILTER
+" AND business_date>=TIMESTAMPADD(DAY,-30,#{p.anchor}) AND business_date<TIMESTAMPADD(DAY,1,#{p.anchor}) GROUP BY customer_code,sku_code),"
+" result AS (SELECT c.customer_code,c.sku_code,c.product_name,c.quantity_unit,c.evidence_at_utc,CASE WHEN "+valid+" THEN 1 ELSE 0 END unit_valid,"
+"CAST(NULL AS DECIMAL(20,6)) observed_quantity,CAST(NULL AS SIGNED) sku_order_count,0 complete_day_count,0 partial_day_count,0 unknown_day_count,"
+"CASE WHEN "+valid+" AND #{p.completeWindow}=1 THEN COALESCE(m.window_quantity,0) END window_quantity,"
+"CASE WHEN "+valid+" AND #{p.completeWindow}=1 THEN COALESCE(m.window_orders,0) END window_sku_order_count,"
+"CASE WHEN "+valid+" AND #{p.complete30}=1 THEN COALESCE(m.sales30_quantity,0) END sales30_quantity,"
+"CASE WHEN "+valid+" AND #{p.salesKnown}=1 THEN m.today_quantity END today_quantity,"
+"st.physical_quantity,st.available_quantity,st.frozen_quantity,st.transit_quantity,st.observed_at_utc inventory_observed_at_utc,"
+"CASE WHEN st.sku_code IS NULL THEN 'UNKNOWN' WHEN st.quality_status='INVALID' THEN 'INVALID' "
+"WHEN st.quality_status<>'VALID' OR st.available_quantity IS NULL OR st.physical_quantity IS NULL OR st.frozen_quantity IS NULL OR st.transit_quantity IS NULL THEN 'UNKNOWN' "
+"WHEN st.available_quantity<0 OR st.physical_quantity<0 OR st.frozen_quantity<0 OR st.transit_quantity<0 THEN 'INVALID' "
+"WHEN #{p.stockFresh}=0 THEN 'STALE' ELSE 'VALID' END inventory_quality,st.quality_reason inventory_reason "
+"FROM evidence c LEFT JOIN metric_sales m ON m.customer_code=c.customer_code AND m.sku_code=c.sku_code "
+"LEFT JOIN ll_selfop_inventory_snapshot st ON st.snapshot_run_id=#{p.inventoryRun} AND st.source_account=#{p.account} "
+"AND st.warehouse_code=#{p.warehouse} AND st.customer_code=c.customer_code AND st.sku_code=c.sku_code AND st.product_quality=0) ";
}
private static String filter(String kind){
if("HOT".equals(kind))return " WHERE window_quantity>0 ";
if("SLOW".equals(kind))return " WHERE window_quantity IS NOT NULL AND inventory_quality='VALID' AND available_quantity>0 "
@@ -119,18 +151,21 @@ public final class AnalyticsReadSql {
public static String count(java.util.Map<String,Object> parameters){
var p=(AnalyticsSqlRequest)parameters.get("p");
// Candidate totals do not scan daily rows for normal lists.
if(!"HOT".equals(p.kind())&&!"SLOW".equals(p.kind()))return typed("SELECT COUNT(*) FROM (SELECT customer_code,sku_code FROM ll_selfop_sku_profile "
return typed("SELECT COUNT(*) FROM (SELECT customer_code,sku_code FROM ll_selfop_sku_profile "
+"WHERE profile_run_id=#{p.salesRun} AND"+KEY_FILTER+" UNION SELECT customer_code,sku_code FROM ll_selfop_sku_profile "
+"WHERE profile_run_id=#{p.inventoryRun} AND"+KEY_FILTER+") candidate_keys",p);
return typed(cte(p.kind())+"SELECT LEAST(COUNT(*),100) FROM result"+filter(p.kind()),p);
}
public static String excluded(java.util.Map<String,Object> parameters){
public static String rankingStats(java.util.Map<String,Object> parameters){
var p=(AnalyticsSqlRequest)parameters.get("p");
String reason="HOT".equals(p.kind())?"CASE WHEN unit_valid=0 THEN 'UNIT_UNKNOWN_OR_CONFLICT' WHEN window_quantity IS NULL THEN 'SALES_WINDOW_UNAVAILABLE' ELSE 'ZERO_OUTBOUND' END":
"CASE WHEN inventory_quality<>'VALID' THEN CONCAT('INVENTORY_',inventory_quality) WHEN available_quantity<=0 THEN 'NO_AVAILABLE_INVENTORY' "
+"WHEN window_quantity IS NULL THEN 'SALES_WINDOW_UNAVAILABLE' ELSE 'OBSERVATION_INSUFFICIENT' END";
return typed(cte(p.kind())+"SELECT "+reason+" reason,COUNT(*) total FROM result WHERE NOT ("+filter(p.kind()).replace(" WHERE ","").strip()+") OR window_quantity IS NULL"
+("SLOW".equals(p.kind())?" OR evidence_at_utc IS NULL":"")+" GROUP BY reason",p);
String eligible=filter(p.kind()).replace(" WHERE ","").strip();
return typed(cte(p.kind())+", classified AS (SELECT CASE WHEN "+eligible+" THEN 1 ELSE 0 END eligible,"
+reason+" reason FROM result), counts AS (SELECT eligible,CASE WHEN eligible=1 THEN NULL ELSE reason END reason,COUNT(*) amount "
+"FROM classified GROUP BY eligible,CASE WHEN eligible=1 THEN NULL ELSE reason END) "
+"SELECT LEAST(SUM(CASE WHEN eligible=1 THEN amount ELSE 0 END) OVER (),100) total,reason,"
+"CASE WHEN eligible=0 THEN amount ELSE 0 END excluded_count FROM counts",p);
}
private static String typed(String sql,AnalyticsSqlRequest p){
String filters=" source_account=#{p.account} AND warehouse_code=#{p.warehouse} ";
@@ -23,6 +23,12 @@ public class SelfopEgressService {
checkAccount(account,budget,now);
SelfopValidationBudget validation=null;
if("VALIDATION".equals(context.executionPurpose())){
if(SelfopCollectionPolicy.warehouseDiscovery(root)){
if(operation!=Operation.WAREHOUSES)throw reject("VALIDATION_DIRECTORY_OPERATION_ONLY");
}else{
String selected=SelfopCollectionPolicy.validationWarehouse(root);
if((operation==Operation.PRODUCT_INVENTORY||operation==Operation.OUTBOUND_DETAIL)&&!Objects.equals(selected,warehouseCode))throw reject("EGRESS_WAREHOUSE_INVALID");
}
validation=validation(root);String limit=validation.limit(now,operation,pageSize,warehouseCode,detailKey);
if(limit!=null){stop(root,validation,limit,false,now);throw new EgressRejectedException(limit,true,null);}
if(operation==Operation.PRODUCT_INVENTORY&&(warehouseCode==null||!config.getWarehouses().containsKey(warehouseCode)))throw reject("EGRESS_WAREHOUSE_INVALID");
@@ -11,6 +11,7 @@ public final class SelfopValidationBudget {
public Set<String> detailKeys=new LinkedHashSet<>();
public Map<String,Integer> inventoryPages=new LinkedHashMap<>();
public static String freeze(LocalDateTime now){return freeze(now,10,20,40,300);}
public static String freezeWarehouseDiscovery(LocalDateTime now){return freeze(now,1,1,1,60);}
public static String freeze(LocalDateTime now,int pages,int details,int http,int seconds){
if(pages<1||details<1||http<1||seconds<1)throw new IllegalArgumentException("VALIDATION_LIMIT_INVALID");
var b=new SelfopValidationBudget();b.maxPages=Math.min(10,pages);b.maxDetailHttp=Math.min(20,details);b.maxDistinctDetails=Math.min(20,details);b.maxHttp=Math.min(40,http);b.startedAtUtc=now;b.deadlineAtUtc=now.plusSeconds(Math.min(300,seconds));return SelfopSupport.json(b);
@@ -34,7 +34,20 @@ public class SelfopCommandService {
private SelfopSyncTask requestChecked(Command c,String operator){
if(!config.isExecutionEnabled())throw new BizException("采集功能未启用,当前环境不允许执行采集或重试");
if(c==null)throw new BizException("请提供采集命令");
input(c.requestKey(),100,"请求标识");input(c.warehouseCode(),64,"仓库编码");
input(c.requestKey(),100,"请求标识");
if("WAREHOUSE_DISCOVERY".equals(c.action())) {
require("VALIDATION".equals(SelfopCollectionPolicy.configuredPurpose(config)),"SELFOP_VALIDATION_ENTRY_REQUIRED");
require(c.warehouseCode()==null&&c.taskNo()==null&&c.sourceFrom()==null&&c.sourceUntil()==null
&&c.customerCode()==null&&c.deliveryNo()==null&&c.reason()==null,"SELFOP_WAREHOUSE_DISCOVERY_PARAMETERS");
var account=taskService.initializeAccount();
Map<String,Object> query=Map.of("action","WAREHOUSE_DISCOVERY","sourceAccount",config.getSourceAccount());
var task=new SelfopSyncTask();task.setSourceAccount(config.getSourceAccount());task.setRequestedBy(operator);
task.setRequestKey("manual:"+c.requestKey());task.setExecutionPurpose("VALIDATION");task.setTaskType("VALIDATION");
task.setTaskRole("ACCOUNT_DISCOVERY");task.setDiscoveryScopeId(account.getId());
task.setQueryScopeJson(json(query));task.setScopeHash(hash(query));task.setRequestHash(hash(Arrays.asList(c,task.getScopeHash())));
return taskService.create(task);
}
input(c.warehouseCode(),64,"仓库编码");
require(config.getWarehouses().containsKey(c.warehouseCode()),"SELFOP_WAREHOUSE_UNCONFIGURED");
if(c.action()==null||!Set.of("RETRY","INVENTORY","BACKFILL","RECHECK","REBUILD","VALIDATION").contains(c.action()))
throw new BizException("不支持此采集操作");
@@ -68,7 +81,7 @@ public class SelfopCommandService {
task.setTaskRole("ACCOUNT_DISCOVERY");task.setDiscoveryScopeId(account.getId());
if("VALIDATION".equals(c.action())){
require(c.sourceFrom()!=null&&c.sourceUntil()!=null&&c.sourceFrom().isBefore(c.sourceUntil())&&!c.sourceUntil().isAfter(c.sourceFrom().plusDays(1)),"SELFOP_QUERY_WINDOW_INVALID");
task.setTaskType("VALIDATION");task.setQueryScopeJson(json(Map.of("startTime",c.sourceFrom().toString(),"endTime",c.sourceUntil().toString(),"warehouses",config.getWarehouses().keySet())));
task.setTaskType("VALIDATION");task.setQueryScopeJson(json(Map.of("action","VALIDATION","startTime",c.sourceFrom().toString(),"endTime",c.sourceUntil().toString(),"warehouses",List.of(c.warehouseCode()))));
}else if("RECHECK".equals(c.action())){identity(c.customerCode(),64);identity(c.deliveryNo(),128);
task.setQueryScopeJson(json(new SelfopDiscoveryService.Query(null,null,c.deliveryNo(),c.customerCode(),null)));
}else{
@@ -99,6 +112,7 @@ public class SelfopCommandService {
Map.entry("SELFOP_SLICE_MAX_ONE_DAY","单次采集时间范围不能超过一天"),
Map.entry("SELFOP_SOURCE_TIME_AMBIGUOUS","来源时间存在时区歧义,请重新选择时间范围"),
Map.entry("SELFOP_SCOPE_MISSING","采集作用域尚未准备就绪,请联系管理员"),
Map.entry("SELFOP_WAREHOUSE_DISCOVERY_PARAMETERS","仓库目录发现仅接受请求标识和操作类型,不接受仓库或采集范围"),
Map.entry("SELFOP_IDEMPOTENCY_CONFLICT","请求标识已用于其他内容,请刷新后重新提交"),
Map.entry("SELFOP_RETRY_LEASE_BUSY","任务正在执行,暂不能重试"),
Map.entry("SELFOP_TASK_ALREADY_SUCCEEDED","任务已成功完成,无需重试"),
@@ -32,7 +32,10 @@ public class SelfopSchedulePlanner {
public void tick(){
if("OFF".equals(config.getCollectionMode()))return;
if("BUSINESS".equals(config.getCollectionMode()))SelfopCollectionPolicy.requireExecution(config,"BUSINESS");
var configured=taskService.initialize();var now=utcNow();
List<SelfopSyncScope> configured;
if("VALIDATION".equals(config.getCollectionMode())){taskService.initializeAccount();configured=List.of();}
else configured=taskService.initialize();
var now=utcNow();
for(var scope:configured){
if(!"BUSINESS".equals(config.getCollectionMode()))break;
if("INVENTORY".equals(scope.getDataFamily()))plan(scope,now);
@@ -30,14 +30,7 @@ public class SelfopTaskService {
public List<SelfopSyncScope> initialize() {
identity(config.getSourceAccount(),64); require(!config.getWarehouses().isEmpty(),"SELFOP_WAREHOUSE_UNCONFIGURED");
var now=utcNow();
var a=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",config.getSourceAccount()));
if(a==null){a=new SelfopOutboundDiscoveryScope();a.setId(IdWorker.getId());a.setSourceAccount(config.getSourceAccount());
a.setNextSequence(1L);a.setNextObservationSequence(1L);a.setFenceToken(0L);
a.setRequestBudget(config.getRequestBudget());a.setTimeBudgetSeconds(config.getTimeBudgetSeconds());
a.setMinRequestIntervalMs(config.getMinRequestIntervalMs());a.setCreatedAt(now);a.setUpdatedAt(now);
require(a.getRequestBudget()>0&&a.getTimeBudgetSeconds()>0&&a.getMinRequestIntervalMs()>0,"SELFOP_BUDGET_INVALID");
try{accounts.insert(a);}catch(DuplicateKeyException ignored){ }
}
initializeAccount();
for(var e:config.getWarehouses().entrySet()) for(String family:List.of("OUTBOUND","INVENTORY")) {
identity(e.getKey(),64);var p=e.getValue();ZoneId.of(p.getBusinessTimezone());
var existing=scopes.selectOne(new QueryWrapper<SelfopSyncScope>().eq("source_account",config.getSourceAccount()).eq("warehouse_code",e.getKey()).eq("data_family",family));
@@ -79,6 +72,23 @@ public class SelfopTaskService {
return scopes.selectList(new QueryWrapper<SelfopSyncScope>().eq("source_account",config.getSourceAccount()).orderByAsc("warehouse_code","data_family"));
}
@Transactional
public SelfopOutboundDiscoveryScope initializeAccount() {
identity(config.getSourceAccount(),64);
var now=utcNow();
var account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",config.getSourceAccount()));
if(account!=null)return account;
account=new SelfopOutboundDiscoveryScope();account.setId(IdWorker.getId());account.setSourceAccount(config.getSourceAccount());
account.setNextSequence(1L);account.setNextObservationSequence(1L);account.setFenceToken(0L);
account.setRequestBudget(config.getRequestBudget());account.setTimeBudgetSeconds(config.getTimeBudgetSeconds());
account.setMinRequestIntervalMs(config.getMinRequestIntervalMs());account.setCreatedAt(now);account.setUpdatedAt(now);
require(account.getRequestBudget()>0&&account.getTimeBudgetSeconds()>0&&account.getMinRequestIntervalMs()>0,"SELFOP_BUDGET_INVALID");
try{accounts.insert(account);}catch(DuplicateKeyException ignored){
account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",config.getSourceAccount()));
require(account!=null,"SELFOP_SCOPE_MISSING");
}
return account;
}
@Transactional
public SelfopSyncTask create(SelfopSyncTask task) {
identity(task.getRequestKey(),128); var existing=tasks.selectOne(new QueryWrapper<SelfopSyncTask>()
.eq("source_account",task.getSourceAccount()).eq("request_key",task.getRequestKey()));
@@ -89,7 +99,12 @@ public class SelfopTaskService {
task.setTaskNo("LXCOL-"+now.atZone(ZoneOffset.UTC).withZoneSameInstant(ZoneId.of("Asia/Shanghai")).toLocalDate().toString().replace("-","")+"-"+UUID.randomUUID().toString().replace("-","").substring(0,16).toUpperCase(Locale.ROOT));
if("VALIDATION".equals(task.getExecutionPurpose())){
require("VALIDATION".equals(task.getTaskType()),"SELFOP_VALIDATION_ENTRY_REQUIRED");
if(task.getValidationRootTaskId()==null){task.setValidationRootTaskId(task.getId());task.setValidationBudgetJson(com.baiyue.lastleg.selfoperated.analytics.request.SelfopValidationBudget.freeze(now));}
if(task.getValidationRootTaskId()==null){task.setValidationRootTaskId(task.getId());
boolean directoryOnly=warehouseDiscovery(task);
if(!directoryOnly)validationWarehouse(task);
task.setValidationBudgetJson(directoryOnly
?com.baiyue.lastleg.selfoperated.analytics.request.SelfopValidationBudget.freezeWarehouseDiscovery(now)
:com.baiyue.lastleg.selfoperated.analytics.request.SelfopValidationBudget.freeze(now));}
task.setApplyState("NOT_APPLICABLE");
}
task.setStatus(task.getStatus()==null?"READY":task.getStatus());
@@ -44,19 +44,30 @@ public class SelfopValidationRunner {
try {
var state=task.getRoutingManifestJson()==null?new Sample():read(task.getRoutingManifestJson(),Sample.class);
Map<?,?> query=read(task.getQueryScopeJson(),Map.class);
var from=LocalDateTime.parse(Objects.toString(query.get("startTime"),""));
var until=LocalDateTime.parse(Objects.toString(query.get("endTime"),""));
require(until.isAfter(from)&&Duration.between(from,until).compareTo(Duration.ofDays(1))<=0,"SELFOP_VALIDATION_WINDOW_INVALID");
boolean directoryOnly=SelfopCollectionPolicy.warehouseDiscovery(task);
String selectedWarehouse=directoryOnly?null:SelfopCollectionPolicy.validationWarehouse(task);
if(!directoryOnly)require(config.getWarehouses().containsKey(selectedWarehouse),"SELFOP_WAREHOUSE_UNCONFIGURED");
var from=directoryOnly?null:LocalDateTime.parse(Objects.toString(query.get("startTime"),""));
var until=directoryOnly?null:LocalDateTime.parse(Objects.toString(query.get("endTime"),""));
if(!directoryOnly)require(until.isAfter(from)&&Duration.between(from,until).compareTo(Duration.ofDays(1))<=0,"SELFOP_VALIDATION_WINDOW_INVALID");
if(!state.warehousesRead) {
leases.renew(lease); var page=request(task,run,lease,()->client.warehouses(context,1,10)); validatePage(page,1);
tx.executeWithoutResult(s->{check(task,run,lease);
Set<String> warehouseCodes=new HashSet<>();
for(var warehouse:page.records()) {
identity(warehouse.warehouseCode(),64);
require(warehouseCodes.add(warehouse.warehouseCode()),"SELFOP_SAMPLE_WAREHOUSE_DUPLICATE");
}
run.setObservedDimensionsJson(json(Map.of("validationWarehouseSamples",page.records(),"truncated",page.pages()>1)));
state.warehousesRead=true;state.pages++;save(task,run,state,"DIRECTORY","PAGE",state.pages,null);
});
}
if(directoryOnly){
require(run.getObservedDimensionsJson()!=null,"SELFOP_WAREHOUSE_DIRECTORY_EVIDENCE_MISSING");
Map<?,?> dimensions=read(run.getObservedDimensionsJson(),Map.class);
finish(task,run,lease,Boolean.TRUE.equals(dimensions.get("truncated"))
?"VALIDATION_WAREHOUSE_DIRECTORY_TRUNCATED":"VALIDATION_WAREHOUSE_DIRECTORY_COMPLETE",false);return;
}
while(!state.directoryRead&&state.outboundPage<=2) {
int current=state.outboundPage;leases.renew(lease);
var page=request(task,run,lease,()->client.discover(context,from,until,current,10));validatePage(page,current);
@@ -64,6 +75,7 @@ public class SelfopValidationRunner {
for(var summary:page.records()) {
require(Objects.equals(task.getSourceAccount(),summary.sourceAccount()),"SELFOP_SAMPLE_ACCOUNT_MISMATCH");
identity(summary.customerCode(),64);identity(summary.warehouseCode(),64);identity(summary.deliveryNo(),128);
if(!selectedWarehouse.equals(summary.warehouseCode()))continue;
String key=hash(List.of(summary.customerCode(),summary.deliveryNo()));
require(state.orders.stream().noneMatch(o->key.equals(hash(List.of(o.customerCode(),o.deliveryNo())))),"SELFOP_SAMPLE_DIRECTORY_DUPLICATE");
state.orders.add(summary);
@@ -75,6 +87,7 @@ public class SelfopValidationRunner {
}
while(state.detailIndex<state.orders.size()) {
var summary=state.orders.get(state.detailIndex);leases.renew(lease);
require(selectedWarehouse.equals(summary.warehouseCode()),"SELFOP_SAMPLE_WAREHOUSE_MISMATCH");
var detail=request(task,run,lease,()->client.detail(context,summary.warehouseCode(),summary.customerCode(),summary.deliveryNo()));
validateOrder(task,summary,detail);
tx.executeWithoutResult(s->{check(task,run,lease);
@@ -85,8 +98,8 @@ public class SelfopValidationRunner {
state.detailIndex++;save(task,run,state,"DETAIL","ORDER",state.detailIndex,(long)state.orders.size());
});
}
// Each configured warehouse is sampled independently, never inferred from directory absence.
for(String warehouse:config.getWarehouses().keySet()) {
// Only the command's frozen warehouse is sampled; other configured warehouses are independent roots.
for(String warehouse:List.of(selectedWarehouse)) {
int current=state.inventoryNextPage.getOrDefault(warehouse,1);
while(current<=2&&!state.inventoryFinished.contains(warehouse)) {
leases.renew(lease);int pageNo=current;
@@ -15,6 +15,9 @@ paths:
BACKFILL 单片不超过一天。RETRY 使用 taskNo 和非空 reason 恢复原分仓逻辑任务,
保留半归并屏障及旧尝试,拒绝验证会话复活、账号共享任务伪装单仓、有效执行租约及上游未齐;
requestKey 同键异载荷拒绝。成功受理只表示排队,不表示采集或发布成功。
WAREHOUSE_DISCOVERY 仅允许 VALIDATION,只接受 requestKey/action,不要求已知仓码;
真实账号根最多一页10条、一次HTTP、60秒,读取后SAMPLE_DONE,不进入出库、详情、库存或正式发布。
普通 VALIDATION 冻结命令指定的单仓,每根独立最多300秒/40HTTP,账号预算、冷却与间隔仍共享。
security:
- bearerAuth: []
x-required-authorities:
@@ -39,7 +42,7 @@ paths:
type: object
properties:
taskNo: { type: string }
warehouseCode: { type: string }
warehouseCode: { type: string, nullable: true, description: 目录发现无仓库主体,其他命令为指定仓码 }
status: { type: string, enum: [ACCEPTED] }
'401': { description: 未认证 }
'403': { description: 非平台身份或无独立写权限 }
@@ -49,11 +52,26 @@ components:
schemas:
SyncCommand:
type: object
required: [requestKey, warehouseCode, action]
required: [requestKey, action]
oneOf:
- properties:
action: { type: string, enum: [WAREHOUSE_DISCOVERY] }
not:
anyOf:
- required: [warehouseCode]
- required: [taskNo]
- required: [sourceFrom]
- required: [sourceUntil]
- required: [customerCode]
- required: [deliveryNo]
- required: [reason]
- required: [warehouseCode]
properties:
action: { type: string, enum: [INVENTORY, BACKFILL, RECHECK, REBUILD, RETRY, VALIDATION] }
properties:
requestKey: { type: string, minLength: 1, maxLength: 100 }
warehouseCode: { type: string, minLength: 1, maxLength: 64 }
action: { type: string, enum: [INVENTORY, BACKFILL, RECHECK, REBUILD, RETRY, VALIDATION] }
action: { type: string, enum: [INVENTORY, BACKFILL, RECHECK, REBUILD, RETRY, VALIDATION, WAREHOUSE_DISCOVERY] }
taskNo: { type: string, minLength: 1, maxLength: 64, description: RETRY 必填,原始分仓业务任务号 }
reason: { type: string, minLength: 1, maxLength: 500, description: RETRY 必填,随操作人和原任务审计保存 }
sourceFrom: { type: string, example: '2026-10-11T00:00:00' }
@@ -30,7 +30,7 @@ class AnalyticsQueryServiceTest {
when(mapper.page(any())).thenReturn(List.of(aggregate(true,"300","VALID",time(D.minusDays(40)))));
when(mapper.inventoryCount(any(),any())).thenReturn(1L);
doAnswer(inv->List.of(inventory(true,"300",inv.getArgument(0,AnalyticsSqlRequest.class).stockFresh(),time(D.minusDays(40)),inv.getArgument(0,AnalyticsSqlRequest.class).complete30()?"150":null))).when(mapper).inventoryPage(any(),any());
when(mapper.excluded(any())).thenReturn(List.of());
when(mapper.rankingStats(any())).thenReturn(List.of(new AnalyticsReadMapper.RankingStat(1L,null,0L)));
when(mapper.profile(11,"demo","W","C","SKU")).thenReturn(profile("PCS","VALID_OUTBOUND",time(D.minusDays(40))));
when(mapper.profile(22,"demo","W","C","SKU")).thenReturn(profile("PCS","INVENTORY_OBSERVATION",time(D.minusDays(35))));
when(mapper.daily(anyLong(),anyString(),anyString(),anyString(),anyString(),any(),any())).thenReturn(List.of());
@@ -211,12 +211,20 @@ class AnalyticsQueryServiceTest {
assertFalse(cap.getValue().completeWindow()); assertFalse(cap.getValue().complete30());
}
@Test void topOneHundredClipsFinalSqlPageAndSkipsBeyondIt(){
when(mapper.count(any())).thenReturn(100L);
when(mapper.rankingStats(any())).thenReturn(List.of(new AnalyticsReadMapper.RankingStat(100L,null,0L)));
service.hotSkus(new AnalyticsQuery("demo","W",null,null,null,null,null,null,30,2,75,11L,22L,SH,IH,NOW));
var cap=ArgumentCaptor.forClass(AnalyticsSqlRequest.class); verify(mapper).page(cap.capture());
assertEquals(75,cap.getValue().offset()); assertEquals(25,cap.getValue().size());
assertTrue(service.hotSkus(new AnalyticsQuery("demo","W",null,null,null,null,null,null,30,3,50,11L,22L,SH,IH,NOW)).records().isEmpty());
verify(mapper,times(1)).page(any());
verify(mapper,never()).count(any());
}
@Test void rankingTotalAndExclusionsUseOneStatisticsQuery(){
when(mapper.rankingStats(any())).thenReturn(List.of(new AnalyticsReadMapper.RankingStat(1L,null,0L),
new AnalyticsReadMapper.RankingStat(1L,"ZERO_OUTBOUND",7L)));
var result=service.hotSkus(metrics(30));assertEquals(1L,result.total());
assertEquals(Map.of("ZERO_OUTBOUND",7L),result.excludedReasons());
verify(mapper).rankingStats(any());verify(mapper).page(any());verify(mapper,never()).count(any());
}
@Test void readRunsWithinConsistentSnapshot(){
var t=AnalyticsQueryService.class.getAnnotation(Transactional.class); assertTrue(t.readOnly()); assertEquals(Isolation.REPEATABLE_READ,t.isolation());
@@ -25,7 +25,7 @@ class AnalyticsReadMapperTest {
}
@Test void allMapperStatementsAreSelectAndProvidersBindTheNamedRecord(){
var c=config(); assertNotNull(AnalyticsReadMapper.class.getAnnotation(Mapper.class));
for(String method:List.of("scopes","publishedRun","page","count","excluded","profile","daily","applyBarriers")){
for(String method:List.of("scopes","publishedRun","page","count","rankingStats","profile","daily","applyBarriers")){
var statement=c.getMappedStatement(NS+method); assertEquals(SqlCommandType.SELECT,statement.getSqlCommandType());
Map<String,Object> params=new HashMap<>(); params.put("p",request("HOT")); params.put("id",11L); params.put("run",11L);
params.put("account","demo"); params.put("warehouse","W"); params.put("customer","OWNER"); params.put("sku","Exact-SKU");
@@ -34,7 +34,10 @@ class AnalyticsReadMapperTest {
}
var b=bound(c,"page",Map.of("p",request("HOT")));
var names=b.getParameterMappings().stream().map(org.apache.ibatis.mapping.ParameterMapping::getProperty).toList();
for(String property:List.of("p.account","p.warehouse","p.customer","p.sku","p.salesRun","p.inventoryRun","p.from","p.until","p.size","p.offset"))assertTrue(names.contains(property),property);
for(String property:List.of("p.account","p.warehouse","p.customer","p.sku","p.salesRun","p.inventoryRun","p.anchor","p.windowFrom","p.size","p.offset"))assertTrue(names.contains(property),property);
var summaryNames=bound(c,"page",Map.of("p",request("SUMMARY"))).getParameterMappings().stream()
.map(org.apache.ibatis.mapping.ParameterMapping::getProperty).toList();
for(String property:List.of("p.from","p.until"))assertTrue(summaryNames.contains(property),property);
assertFalse(b.getSql().contains("#{"));
assertEquals(AnalyticsReadRows.Aggregate.class,c.getMappedStatement(NS+"page").getResultMaps().get(0).getType());
}
@@ -56,6 +59,7 @@ class AnalyticsReadMapperTest {
assertTrue(page.contains("d.sales_run_id=?")); assertTrue(page.contains("d.business_date>=CAST(? AS DATE) AND d.business_date<CAST(? AS DATE)"));
assertTrue(page.contains("GROUP BY b.customer_code,b.sku_code")); assertFalse(page.contains("business_date>=DATE_SUB"));
assertFalse(page.contains("ll_selfop_inventory_snapshot"));
assertFalse(page.contains("metric_sales"));assertFalse(page.contains("FROM bounds WHERE 1=0"));
assertTrue(page.endsWith("ORDER BY observed_quantity IS NULL,observed_quantity DESC,sku_code ASC,customer_code ASC LIMIT ? OFFSET ?"));
assertTrue(page.indexOf("GROUP BY b.customer_code,b.sku_code")<page.lastIndexOf("LIMIT")); immutableOnly(page);
}
@@ -77,9 +81,23 @@ class AnalyticsReadMapperTest {
assertTrue(slow.contains("evidence_at_utc<=CONVERT_TZ"));
assertTrue(slow.contains("ORDER BY window_quantity ASC,available_quantity DESC,sku_code ASC,customer_code ASC LIMIT"));
for(String kind:List.of("HOT","SLOW")){
String count=bound(c,"count",Map.of("p",request(kind))).getSql(); assertTrue(count.contains("SELECT LEAST(COUNT(*),100) FROM result")); assertFalse(count.contains(" LIMIT "));
String stats=bound(c,"rankingStats",Map.of("p",request(kind))).getSql();
assertTrue(stats.contains("LEAST(SUM(CASE WHEN eligible=1 THEN amount ELSE 0 END) OVER (),100) total"));
assertTrue(stats.contains("excluded_count"));assertFalse(stats.contains(" LIMIT "));
assertFalse(stats.contains("ranged AS"));assertFalse(stats.contains("TIMESTAMPDIFF"));
}
}
@Test void inventoryHistoryOnlyEnrichesQualifiedPageAndNeverCountsOrOwnerTotals(){
var c=config();var p=request("INVENTORY");var o=new AnalyticsInventoryOptions(null,"COVERAGE_GT_THRESHOLD","PHYSICAL_DESC",90);
String page=bound(c,"inventoryPage",Map.of("p",p,"o",o)).getSql();
assertTrue(page.contains("paged AS (SELECT * FROM result WHERE above_threshold=1 ORDER BY"));
assertTrue(page.indexOf("LIMIT ? OFFSET ?")<page.indexOf("last_sales AS"));
assertTrue(page.contains("JOIN paged k ON k.customer_code=d.customer_code AND k.sku_code=d.sku_code"));
assertTrue(page.contains("d.sales_run_id=? AND d.source_account=? AND d.warehouse_code=?"));
assertFalse(bound(c,"inventoryCount",Map.of("p",p,"o",o)).getSql().contains("last_sales"));
assertFalse(bound(c,"ownerTotals",Map.of("p",p,"o",o)).getSql().contains("last_sales"));
immutableOnly(page);
}
@Test void observationAndUnitsRequireImmutableProofAndPcsNotFirstSeenOrStockPlaceholders(){
String sql=bound(config(),"page",Map.of("p",request("INVENTORY"))).getSql();
assertTrue(sql.contains("s.evidence_type='VALID_OUTBOUND' AND s.evidence_run_id IS NOT NULL"));
@@ -163,6 +163,42 @@ public class AnalyticsSqlSemanticTest {
var totals=totals(request("INVENTORY",D.minusDays(30),D,"C",null,utc(D,8),0,20),new AnalyticsInventoryOptions(null,null,null,30));
assertDecimal("12",(BigDecimal)totals.get("owner_outbound_known"));assertEquals(0L,((Number)totals.get("owner_outbound_unknown")).longValue());
}
@Test void rankingStatisticsPreserveFullEligibilityTopCapExclusionsAndSharedSkuOwners() throws Exception {
for(int i=0;i<110;i++){
String sku=String.format("SKU-%03d",i);profile(11,"C",sku,"VALID_OUTBOUND",utc(D.minusDays(40),0),"PCS");
profile(22,"C",sku,"INVENTORY_OBSERVATION",utc(D.minusDays(40),0),"PCS");
daily(11,"W","C",sku,D.minusDays(1),"1");stock("C",sku,0,"10");
}
profile(11,"B","SKU-000","VALID_OUTBOUND",utc(D.minusDays(40),0),"PCS");
profile(22,"B","SKU-000","INVENTORY_OBSERVATION",utc(D.minusDays(40),0),"PCS");
daily(11,"W","B","SKU-000",D.minusDays(1),"2");stock("B","SKU-000",0,"10");
profile(22,"C","ZERO","INVENTORY_OBSERVATION",utc(D.minusDays(40),0),"PCS");stock("C","ZERO",0,"0");
profile(22,"C","NEW","INVENTORY_OBSERVATION",utc(D.minusDays(1),0),"PCS");stock("C","NEW",0,"10");
profile(22,"C","UNIT","INVENTORY_OBSERVATION",utc(D.minusDays(40),0),null);stock("C","UNIT",0,"10");
var hot=request("HOT",D.minusDays(30),D,null,null,utc(D,8),0,20);
var hotStats=rankingStats(hot);assertTrue(hotStats.stream().allMatch(s->s.total()==100L));
assertEquals(2L,hotStats.stream().filter(s->"ZERO_OUTBOUND".equals(s.reason())).findFirst().orElseThrow().excludedCount());
assertEquals(1L,hotStats.stream().filter(s->"UNIT_UNKNOWN_OR_CONFLICT".equals(s.reason())).findFirst().orElseThrow().excludedCount());
assertEquals("B",page(hot).get(0).customerCode());
var slowStats=rankingStats(request("SLOW",D.minusDays(30),D,null,null,utc(D,8),0,20));
assertTrue(slowStats.stream().allMatch(s->s.total()==100L));
for(String reason:List.of("NO_AVAILABLE_INVENTORY","OBSERVATION_INSUFFICIENT","SALES_WINDOW_UNAVAILABLE"))
assertEquals(1L,slowStats.stream().filter(s->reason.equals(s.reason())).findFirst().orElseThrow().excludedCount());
}
@Test void inventoryPageKeepsHistoricLastDateForOnlySelectedQualifiedKeys() throws Exception {
for(String sku:List.of("OLD-1","OLD-2","RECENT")){
profile(22,"C",sku,"INVENTORY_OBSERVATION",utc(D.minusDays(90),0),"PCS");stock("C",sku,0,"10");
}
daily(11,"W","C","OLD-1",D.minusDays(35),"4");daily(11,"W","C","OLD-2",D.minusDays(60),"8");
daily(11,"W","C","RECENT",D.minusDays(1),"2");
var first=inventory(null,30,"PHYSICAL_DESC",0,1);assertEquals("OLD-1",first.get(0).skuCode());
assertEquals(D.minusDays(35),first.get(0).lastOutbound());
var second=inventory(null,30,"PHYSICAL_DESC",1,1);assertEquals("OLD-2",second.get(0).skuCode());
assertEquals(D.minusDays(60),second.get(0).lastOutbound());
}
@SuppressWarnings("unchecked") private List<AnalyticsReadMapper.RankingStat> rankingStats(AnalyticsSqlRequest p){
return (List<AnalyticsReadMapper.RankingStat>)(List<?>)literalQuery("rankingStats",p,new AnalyticsInventoryOptions(null,null,null,null));
}
private static void assertDecimal(String expected,BigDecimal actual){assertNotNull(actual);assertEquals(0,new BigDecimal(expected).compareTo(actual));}
@SuppressWarnings("unchecked")
private List<AnalyticsReadRows.Inventory> inventory(String attention,Integer threshold,String sort,long offset,int size){
@@ -19,6 +19,7 @@ class SelfopEgressServiceTest {
cfg.setSourceAccount("synthetic");cfg.setCollectionMode("VALIDATION");cfg.setExclusiveGrantId("grant");cfg.setExclusiveGrantExpiresAtUtc(now().plusDays(2));cfg.setExclusiveHandoffVerified(true);cfg.setPriorEgressClosed(true);cfg.setInflightDrained(true);cfg.setTransferredBudgetVerified(true);cfg.setTransferredBudgetDate(now().plusHours(8).toLocalDate());cfg.setDailyRequestBudget(100);cfg.getWarehouses().put("A",new SelfopAnalyticsProperties.Warehouse());cfg.getWarehouses().put("B",new SelfopAnalyticsProperties.Warehouse());
account.setId(10L);account.setSourceAccount("synthetic");account.setActiveTaskId(1L);account.setFenceToken(7L);account.setLeaseUntilUtc(now().plusDays(2));
task.setId(1L);task.setDiscoveryScopeId(10L);task.setSourceAccount("synthetic");task.setStatus("RUNNING");task.setTaskRole("ACCOUNT_DISCOVERY");task.setExecutionPurpose("VALIDATION");task.setValidationRootTaskId(1L);task.setValidationBudgetJson(SelfopValidationBudget.freeze(now()));
task.setTaskType("VALIDATION");task.setQueryScopeJson("{\"action\":\"VALIDATION\",\"warehouses\":[\"A\"]}");
run.setId(2L);run.setTaskId(1L);run.setSourceAccount("synthetic");run.setStatus("FETCHING");run.setExecutionPurpose("VALIDATION");run.setFenceToken(7L);
when(mapper.lockAccount("synthetic")).thenReturn(account);when(mapper.lockTask(1L)).thenReturn(task);when(mapper.lockRun(2L)).thenReturn(run);when(mapper.saveAccount(any())).thenReturn(1);when(mapper.saveRoot(any())).thenReturn(1);
}
@@ -59,9 +60,12 @@ class SelfopEgressServiceTest {
assertThatThrownBy(()->new SelfopEgressService(mapper,cfg,clock).reserve(context(),Operation.OUTBOUND_QUERY,1,null,null,INTERNAL)).hasMessage("VALIDATION_PAGE_LIMIT").satisfies(e->assertThat(((EgressRejectedException)e).sampleDone()).isTrue());
assertThat(task.getStatus()).isEqualTo("SAMPLE_DONE");assertThat(rootBudget().http).isEqualTo(10);verify(mapper).terminateValidationRuns(eq(1L),eq("SAMPLE_DONE"),eq("VALIDATION_PAGE_LIMIT"),any());
}
@Test void inventoryTwoPageLimitPerWarehouseSharesGlobalBudget(){
success(reserve(Operation.PRODUCT_INVENTORY,1,"A",null),Operation.PRODUCT_INVENTORY);success(reserve(Operation.PRODUCT_INVENTORY,1,"B",null),Operation.PRODUCT_INVENTORY);success(reserve(Operation.PRODUCT_INVENTORY,1,"A",null),Operation.PRODUCT_INVENTORY);
assertThatThrownBy(()->reserve(Operation.PRODUCT_INVENTORY,1,"A",null)).hasMessage("VALIDATION_INVENTORY_LIMIT");assertThat(rootBudget().inventoryPages).containsEntry("A",2).containsEntry("B",1);assertThat(rootBudget().http).isEqualTo(3);
@Test void inventoryTwoPageLimitCannotSampleAnotherConfiguredWarehouse(){
success(reserve(Operation.PRODUCT_INVENTORY,1,"A",null),Operation.PRODUCT_INVENTORY);
assertThatThrownBy(()->reserve(Operation.PRODUCT_INVENTORY,1,"B",null)).hasMessage("EGRESS_WAREHOUSE_INVALID");
assertThatThrownBy(()->reserve(Operation.OUTBOUND_DETAIL,0,"B","C/DO")).hasMessage("EGRESS_WAREHOUSE_INVALID");
success(reserve(Operation.PRODUCT_INVENTORY,1,"A",null),Operation.PRODUCT_INVENTORY);
assertThatThrownBy(()->reserve(Operation.PRODUCT_INVENTORY,1,"A",null)).hasMessage("VALIDATION_INVENTORY_LIMIT");assertThat(rootBudget().inventoryPages).containsEntry("A",2).doesNotContainKey("B");assertThat(rootBudget().http).isEqualTo(2);
}
@Test void repeatedDetailUsesHttpLimitEvenForSameIdentity(){for(int i=0;i<20;i++)success(reserve(Operation.OUTBOUND_DETAIL,0,"A","C/DO"),Operation.OUTBOUND_DETAIL);assertThatThrownBy(()->reserve(Operation.OUTBOUND_DETAIL,0,"A","C/DO")).hasMessage("VALIDATION_DETAIL_HTTP_LIMIT");assertThat(rootBudget().detailKeys).hasSize(1);assertThat(rootBudget().detailHttp).isEqualTo(20);}
@Test void fiveMinuteDeadlineTerminalAndFrozenLimitsCannotIncrease(){clock.advance(300);assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("VALIDATION_DEADLINE");assertThat(task.getStatus()).isEqualTo("SAMPLE_DONE");var b=rootBudget();b.maxHttp=41;assertThatThrownBy(b::validate).hasMessage("VALIDATION_BUDGET_INVALID");}
@@ -71,7 +75,35 @@ class SelfopEgressServiceTest {
@Test void actualStartIntervalUsesAuthorizationTimeAndReturnsQueueDeadline(){var c=reserve(Operation.WAREHOUSES,1,null,null);clock.advance(4);service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER));service.complete(new CompletionRequest(c,Operation.WAREHOUSES,PROVIDER,1,null,200,200,false,null));clock.advance(1);assertThatThrownBy(()->reserve(Operation.OUTBOUND_QUERY,1,null,null)).hasMessage("ACCOUNT_INTERVAL_WAIT").satisfies(e->assertThat(((EgressRejectedException)e).waitUntilUtc()).isEqualTo(now().plusSeconds(4)));}
@Test void obsoleteExecutionFenceCannotReserveOrConsume(){run.setFenceToken(6L);assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EGRESS_FENCE_LOST");run.setFenceToken(7L);var c=reserve(Operation.WAREHOUSES,1,null,null);account.setFenceToken(8L);assertThatThrownBy(()->service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER))).hasMessage("EGRESS_FENCE_LOST");assertThat(budget().reservedHttp).isEqualTo(1);}
@Test void totalHttpCapCountsEveryOperationAndPreparedReservation(){var b=rootBudget();b.http=40;task.setValidationBudgetJson(SelfopSupport.json(b));assertThatThrownBy(()->reserve(Operation.OUTBOUND_DETAIL,0,"A","C/DO")).hasMessage("VALIDATION_HTTP_LIMIT");assertThat(task.getStatus()).isEqualTo("SAMPLE_DONE");}
@Test void warehouseRequestsShareAccountBudgetAndRejectScopeMismatch(){task.setTaskRole("WAREHOUSE_SYNC");task.setScopeId(100L);task.setWarehouseCode("A");run.setScopeId(100L);run.setWarehouseCode("A");when(mapper.warehouseFence(eq(100L),eq(7L),eq(1L),any())).thenReturn(1);var c=reserve(Operation.PRODUCT_INVENTORY,1,"A",null);success(c,Operation.PRODUCT_INVENTORY);assertThat(budget().reservedHttp).isEqualTo(1);run.setWarehouseCode("B");assertThatThrownBy(()->reserve(Operation.PRODUCT_INVENTORY,1,"A",null)).hasMessage("EGRESS_FENCE_LOST");}
@Test void warehouseChildRequestsUseOriginalAccountRootBudgetAndRejectScopeMismatch(){
var child=new SelfopSyncTask();child.setId(3L);child.setValidationRootTaskId(1L);child.setTaskRole("WAREHOUSE_SYNC");child.setExecutionPurpose("VALIDATION");child.setSourceAccount("synthetic");child.setStatus("RUNNING");child.setScopeId(100L);child.setWarehouseCode("A");
var attempt=new SelfopSyncRun();attempt.setId(4L);attempt.setTaskId(3L);attempt.setSourceAccount("synthetic");attempt.setExecutionPurpose("VALIDATION");attempt.setStatus("FETCHING");attempt.setFenceToken(7L);attempt.setScopeId(100L);attempt.setWarehouseCode("A");
when(mapper.lockTask(3L)).thenReturn(child);when(mapper.lockRun(4L)).thenReturn(attempt);when(mapper.warehouseFence(eq(100L),eq(7L),eq(3L),any())).thenReturn(1);
var childContext=new SelfopRequestContext("synthetic","VALIDATION",1L,"selfop-1","grant",3L,4L);
var c=service.reserve(childContext,Operation.PRODUCT_INVENTORY,1,"A",null,INTERNAL);success(c,Operation.PRODUCT_INVENTORY);assertThat(budget().reservedHttp).isEqualTo(1);
attempt.setWarehouseCode("B");assertThatThrownBy(()->service.reserve(childContext,Operation.PRODUCT_INVENTORY,1,"A",null,INTERNAL)).hasMessage("EGRESS_FENCE_LOST");
}
@Test void directoryRootOnlyAllowsWarehouseOperationAndOneHttpWithinSixtySeconds(){
task.setQueryScopeJson("{\"action\":\"WAREHOUSE_DISCOVERY\"}");task.setValidationBudgetJson(SelfopValidationBudget.freezeWarehouseDiscovery(now()));
assertThatThrownBy(()->reserve(Operation.OUTBOUND_QUERY,10,null,null)).hasMessage("VALIDATION_DIRECTORY_OPERATION_ONLY");
assertThatThrownBy(()->reserve(Operation.PRODUCT_INVENTORY,10,"A",null)).hasMessage("VALIDATION_DIRECTORY_OPERATION_ONLY");
assertThat(budget().reservedHttp).isZero();
assertThat(rootBudget().deadlineAtUtc).isEqualTo(rootBudget().startedAtUtc.plusSeconds(60));
success(reserve(Operation.WAREHOUSES,10,null,null),Operation.WAREHOUSES);
assertThatThrownBy(()->reserve(Operation.WAREHOUSES,10,null,null)).hasMessage("VALIDATION_HTTP_LIMIT");
assertThat(budget().reservedHttp).isEqualTo(1);assertThat(budget().confirmedHttp).isEqualTo(1);assertThat(rootBudget().pages).isEqualTo(1);
}
@Test void separateWarehouseRootsKeepTheirOwnFiveMinuteBudgetsAndSameAccountConsumption(){
success(reserve(Operation.PRODUCT_INVENTORY,10,"A",null),Operation.PRODUCT_INVENTORY);service.finishSample(1L,"VALIDATION_SAMPLE_SUFFICIENT");
var bRoot=new SelfopSyncTask();bRoot.setId(3L);bRoot.setValidationRootTaskId(3L);bRoot.setTaskRole("ACCOUNT_DISCOVERY");bRoot.setTaskType("VALIDATION");bRoot.setExecutionPurpose("VALIDATION");bRoot.setSourceAccount("synthetic");bRoot.setDiscoveryScopeId(10L);bRoot.setStatus("RUNNING");bRoot.setQueryScopeJson("{\"action\":\"VALIDATION\",\"warehouses\":[\"B\"]}");bRoot.setValidationBudgetJson(SelfopValidationBudget.freeze(now()));
var bRun=new SelfopSyncRun();bRun.setId(4L);bRun.setTaskId(3L);bRun.setSourceAccount("synthetic");bRun.setExecutionPurpose("VALIDATION");bRun.setStatus("FETCHING");bRun.setFenceToken(7L);
when(mapper.lockTask(3L)).thenReturn(bRoot);when(mapper.lockRun(4L)).thenReturn(bRun);
var c=service.reserve(new SelfopRequestContext("synthetic","VALIDATION",3L,"selfop-3","grant",3L,4L),Operation.PRODUCT_INVENTORY,10,"B",null,INTERNAL);success(c,Operation.PRODUCT_INVENTORY);
assertThat(budget().reservedHttp).isEqualTo(2);assertThat(budget().confirmedHttp).isEqualTo(2);
var second=SelfopSupport.read(bRoot.getValidationBudgetJson(),SelfopValidationBudget.class);
assertThat(rootBudget().http).isEqualTo(1);assertThat(second.http).isEqualTo(1);assertThat(second.maxHttp).isEqualTo(40);
assertThat(second.deadlineAtUtc).isEqualTo(second.startedAtUtc.plusSeconds(300));
}
@Test void unknownGrantTransferCooldownIsPreservedAndExpiredGrantCannotIssue(){cfg.setTransferredCooldownUntilUtc(now().plusSeconds(100));assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("ACCOUNT_COOLDOWN_WAIT");cfg.setExclusiveGrantExpiresAtUtc(now().minusSeconds(1));assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EXCLUSIVE_HANDOFF_UNCONFIRMED");}
@Test void persistedDailyCapCannotBeRaisedMiddayToBypassBudget(){cfg.setDailyRequestBudget(1);success(reserve(Operation.WAREHOUSES,1,null,null),Operation.WAREHOUSES);cfg.setDailyRequestBudget(100);assertThatThrownBy(()->reserve(Operation.OUTBOUND_QUERY,1,null,null)).hasMessage("ACCOUNT_DAILY_BUDGET");assertThat(budget().dailyLimit).isEqualTo(1);}
@Test void oldAttemptCannotConsumeAfterNextPreparedRequest(){var c=reserve(Operation.WAREHOUSES,1,null,null);success(c,Operation.WAREHOUSES);reserve(Operation.OUTBOUND_QUERY,1,null,null);assertThatThrownBy(()->service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER))).hasMessage("EGRESS_PERMIT_BINDING_INVALID");}
@@ -5,6 +5,8 @@ import com.baiyue.lastleg.selfoperated.analytics.model.*;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.read;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;
@@ -90,6 +92,38 @@ class SelfopCommandServiceTest {
assertThatThrownBy(()->service.request(c,"42")).hasMessage("当前采集模式与所选操作不一致,请使用对应入口");
verify(taskService,never()).create(any(SelfopSyncTask.class));
}
@Test void warehouseDiscoveryCreatesAccountRootWithoutAnyWarehouseOrTimeWindow() {
config.setCollectionMode("VALIDATION");config.setSourceAccount("synthetic-account");
var account=new SelfopOutboundDiscoveryScope();account.setId(20L);
when(taskService.initializeAccount()).thenReturn(account);
when(taskService.create(any(SelfopSyncTask.class))).thenAnswer(call->call.getArgument(0));
var task=service.request(new SelfopCommandService.Command("directory",null,"WAREHOUSE_DISCOVERY",null,null,null,null,null,null),"42");
assertThat(task.getTaskRole()).isEqualTo("ACCOUNT_DISCOVERY");assertThat(task.getTaskType()).isEqualTo("VALIDATION");
assertThat(task.getExecutionPurpose()).isEqualTo("VALIDATION");assertThat(task.getDiscoveryScopeId()).isEqualTo(20L);
assertThat(task.getWarehouseCode()).isNull();assertThat(task.getScopeId()).isNull();assertThat(task.getRequestedBy()).isEqualTo("42");
assertThat(read(task.getQueryScopeJson(),Map.class)).containsEntry("action","WAREHOUSE_DISCOVERY").doesNotContainKeys("warehouses","startTime","endTime");
assertThat(task.getScopeHash()).matches("[a-f0-9]{64}");
verify(taskService,never()).initialize();verifyNoInteractions(scopes,accounts,tasks);
}
@Test void directoryDiscoveryRejectsBusinessModeAndAnySuppliedWarehouseBeforeInitializing() {
configure();
var c=new SelfopCommandService.Command("directory",null,"WAREHOUSE_DISCOVERY",null,null,null,null,null,null);
assertThatThrownBy(()->service.request(c,"42")).hasMessage("当前采集模式与所选操作不一致,请使用对应入口");
config.setCollectionMode("VALIDATION");
assertThatThrownBy(()->service.request(new SelfopCommandService.Command("directory","WH-A","WAREHOUSE_DISCOVERY",null,null,null,null,null,null),"42"))
.hasMessage("仓库目录发现仅接受请求标识和操作类型,不接受仓库或采集范围");
verifyNoInteractions(taskService,scopes,accounts,tasks);
}
@Test void ordinaryValidationFreezesOnlySelectedWarehouseDespiteOtherConfiguredWarehouse() {
configure();config.setCollectionMode("VALIDATION");config.getWarehouses().put("WH-B",new SelfopAnalyticsProperties.Warehouse());
var scope=new SelfopSyncScope();scope.setId(10L);scope.setScopeHash("scope");
var account=new SelfopOutboundDiscoveryScope();account.setId(20L);
when(scopes.selectOne(any())).thenReturn(scope);when(accounts.selectOne(any())).thenReturn(account);
when(taskService.create(any(SelfopSyncTask.class))).thenAnswer(call->call.getArgument(0));
var from=LocalDateTime.of(2026,10,12,8,0);
var task=service.request(new SelfopCommandService.Command("sample-a","WH-A","VALIDATION",null,from,from.plusMinutes(5),null,null,null),"42");
assertThat(read(task.getQueryScopeJson(),Map.class)).containsEntry("action","VALIDATION").containsEntry("warehouses",List.of("WH-A"));
}
private SelfopCommandService.Command retry() {
return new SelfopCommandService.Command("retry-key", "WH-A", "RETRY", "LXCOL-20261012-00000001", null, null, null, null, "offline retry");
}
@@ -32,6 +32,14 @@ class SelfopSchedulePlannerTest {
planner.tick();
verifyNoInteractions(taskService, execution, scopes, accounts, tasks, orders, locks, tx);
}
@Test void validationWithoutConfiguredWarehouseOnlyInitializesAccountAndExecutesDurableCommand() {
config.setCollectionMode("VALIDATION");
var due=new SelfopSyncTask();due.setId(101L);due.setTaskType("VALIDATION");due.setExecutionPurpose("VALIDATION");
when(taskService.due(10)).thenReturn(List.of(due));
planner.tick();
verify(taskService).initializeAccount();verify(taskService,never()).initialize();verify(taskService,never()).create(any(SelfopSyncTask.class));
verify(execution).execute(101L);verifyNoInteractions(scopes,accounts,tasks,orders,locks,tx);
}
@Test void unknownSourceTimezoneStillPlansBothInventoriesAndExecutesDurableDueWork() {
config.setCollectionMode("BUSINESS"); config.setEnvironment("production");
@@ -26,6 +26,51 @@ class SelfopTaskServiceTest {
@BeforeEach void configureBusiness() {config.setCollectionMode("BUSINESS");config.setEnvironment("production");}
@Test void accountInitializationNeedsNoWarehouseAndNeverTouchesWarehouseScopes() {
config.setCollectionMode("VALIDATION");config.setSourceAccount("synthetic-account");
var account=service.initializeAccount();
assertThat(account.getSourceAccount()).isEqualTo("synthetic-account");assertThat(account.getNextSequence()).isEqualTo(1L);
assertThat(account.getApiBudgetJson()).isNull();assertThat(account.getApiCooldownUntilUtc()).isNull();
verify(accounts).insert(account);verifyNoInteractions(scopes,tasks,runs,locks,leases);
}
@Test void existingAccountInitializationPreservesConsumedBudgetPauseAndCooldown() {
config.setCollectionMode("VALIDATION");config.setSourceAccount("synthetic-account");
var account=new SelfopOutboundDiscoveryScope();account.setId(20L);account.setSourceAccount("synthetic-account");
account.setApiBudgetJson("{\"reservedHttp\":7}");account.setApiPauseReason("EGRESS_REQUIRES_REVIEW");
account.setApiCooldownUntilUtc(utcNow().plusMinutes(10));
when(accounts.selectOne(any())).thenReturn(account);
assertThat(service.initializeAccount()).isSameAs(account);
assertThat(account.getApiBudgetJson()).isEqualTo("{\"reservedHttp\":7}");assertThat(account.getApiPauseReason()).isEqualTo("EGRESS_REQUIRES_REVIEW");
verify(accounts,never()).insert(any(SelfopOutboundDiscoveryScope.class));verify(accounts,never()).updateById(any(SelfopOutboundDiscoveryScope.class));
verifyNoInteractions(scopes,tasks,runs,locks,leases);
}
@Test void newDirectoryRootFreezesOnePageOneHttpSixtySecondsWithoutFakeWarehouse() {
config.setCollectionMode("VALIDATION");config.setSourceAccount("synthetic-account");
var account=new SelfopOutboundDiscoveryScope();account.setId(20L);account.setNextSequence(1L);account.setNextObservationSequence(1L);
when(locks.lockDiscovery(20L)).thenReturn(account);
var task=new SelfopSyncTask();task.setSourceAccount("synthetic-account");task.setExecutionPurpose("VALIDATION");
task.setTaskType("VALIDATION");task.setTaskRole("ACCOUNT_DISCOVERY");task.setDiscoveryScopeId(20L);
task.setRequestKey("directory");task.setRequestHash("directory-hash");task.setScopeHash("scope-hash");
task.setQueryScopeJson("{\"action\":\"WAREHOUSE_DISCOVERY\",\"sourceAccount\":\"synthetic-account\"}");
service.create(task);
assertThat(task.getValidationRootTaskId()).isEqualTo(task.getId());assertThat(task.getWarehouseCode()).isNull();assertThat(task.getScopeId()).isNull();
var budget=read(task.getValidationBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopValidationBudget.class);
assertThat(budget.maxPages).isEqualTo(1);assertThat(budget.maxHttp).isEqualTo(1);assertThat(budget.maxPageSize).isEqualTo(10);
assertThat(budget.deadlineAtUtc).isEqualTo(budget.startedAtUtc.plusSeconds(60));assertThat(task.getApplyState()).isEqualTo("NOT_APPLICABLE");
verify(tasks).insert(task);verifyNoInteractions(scopes,runs,leases);
}
@Test void singleWarehouseRootKeepsFullFiveMinuteFortyHttpLimit() {
config.setCollectionMode("VALIDATION");config.setSourceAccount("synthetic-account");
var account=new SelfopOutboundDiscoveryScope();account.setId(20L);account.setNextSequence(1L);account.setNextObservationSequence(1L);
when(locks.lockDiscovery(20L)).thenReturn(account);
var task=new SelfopSyncTask();task.setSourceAccount("synthetic-account");task.setExecutionPurpose("VALIDATION");task.setTaskType("VALIDATION");
task.setTaskRole("ACCOUNT_DISCOVERY");task.setDiscoveryScopeId(20L);task.setRequestKey("sample");task.setRequestHash("sample-hash");
task.setQueryScopeJson("{\"action\":\"VALIDATION\",\"warehouses\":[\"WH-A\"]}");
service.create(task);
var budget=read(task.getValidationBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopValidationBudget.class);
assertThat(budget.maxHttp).isEqualTo(40);assertThat(budget.deadlineAtUtc).isEqualTo(budget.startedAtUtc.plusSeconds(300));
}
@Test void beginFixesTopLevelTimezoneStalenessAndInitialWindowInVersionSnapshot() {
var task = task("READY");
var scope = scope("OUTBOUND");
@@ -37,8 +37,8 @@ class SelfopValidationRunnerTest {
config.setCollectionMode("VALIDATION");config.setSourceAccount("demo");config.setExclusiveGrantId("test-grant");
config.getWarehouses().put("A",new SelfopAnalyticsProperties.Warehouse());config.getWarehouses().put("B",new SelfopAnalyticsProperties.Warehouse());
task=new SelfopSyncTask();task.setId(1L);task.setValidationRootTaskId(1L);task.setTaskRole("ACCOUNT_DISCOVERY");task.setExecutionPurpose("VALIDATION");
task.setSourceAccount("demo");task.setDiscoveryScopeId(3L);task.setAttemptCount(1);task.setStatus("RUNNING");
task.setQueryScopeJson("{\"startTime\":\"2026-10-12T08:00:00\",\"endTime\":\"2026-10-12T08:10:00\"}");
task.setSourceAccount("demo");task.setDiscoveryScopeId(3L);task.setAttemptCount(1);task.setStatus("RUNNING");task.setTaskType("VALIDATION");
task.setQueryScopeJson("{\"action\":\"VALIDATION\",\"warehouses\":[\"A\"],\"startTime\":\"2026-10-12T08:00:00\",\"endTime\":\"2026-10-12T08:10:00\"}");
task.setValidationBudgetJson(SelfopValidationBudget.freeze(utcNow()));
run=new SelfopSyncRun();run.setId(2L);run.setTaskId(1L);run.setSourceAccount("demo");run.setExecutionPurpose("VALIDATION");run.setAttemptNo(1);run.setFenceToken(8L);run.setStatus("FETCHING");
when(locks.lockTask(1)).thenReturn(task);when(locks.lockRun(2)).thenReturn(run);when(tasks.selectById(1L)).thenReturn(task);
@@ -52,7 +52,7 @@ class SelfopValidationRunnerTest {
private SelfopPayloads.Stock stock(String warehouse,String sku,BigDecimal qty){return new SelfopPayloads.Stock("demo","customer",warehouse,sku,0,"Name",qty,qty,BigDecimal.ZERO,BigDecimal.ZERO,null);}
private LingxingAnalyticsClient.Summary summary(String number){return new LingxingAnalyticsClient.Summary("demo","customer","A",number,"source",100,null,"2026-10-12 08:02:00");}
@Test void truncatedTwoWarehouseSampleOnlyStagesAndDoesNotClaimCompleteCoverage() {
@Test void selectedWarehouseSampleOnlyStagesItsOwnRowsAndDoesNotClaimCompleteCoverage() {
when(client.inventory(any(SelfopRequestContext.class),eq("A"),eq(1),eq(10))).thenReturn(page(1,30,3,List.of(stock("A","one",BigDecimal.ZERO))));
when(client.inventory(any(SelfopRequestContext.class),eq("A"),eq(2),eq(10))).thenReturn(page(2,30,3,List.of(stock("A","two",null))));
when(client.inventory(any(SelfopRequestContext.class),eq("B"),eq(1),eq(10))).thenReturn(page(1,30,3,List.of(stock("B","one",BigDecimal.ONE))));
@@ -60,8 +60,10 @@ class SelfopValidationRunnerTest {
runner.run(task,run,lease);
assertEquals("SAMPLE_DONE",run.getStatus());assertEquals("UNKNOWN",run.getCoverageStatus());assertNull(run.getPublishedAtUtc());assertNull(run.getCoveredUntilUtc());
verify(client,never()).inventory(any(SelfopRequestContext.class),anyString(),eq(3),anyInt());
var capture=ArgumentCaptor.forClass(SelfopSyncStage.class);verify(stages,times(4)).insert(capture.capture());
var capture=ArgumentCaptor.forClass(SelfopSyncStage.class);verify(stages,times(2)).insert(capture.capture());
assertTrue(capture.getAllValues().stream().allMatch(s->s.getObjectType().equals("PRODUCT_STOCK")&&s.getObservationSequence()==null));
assertTrue(capture.getAllValues().stream().allMatch(s->"A".equals(s.getWarehouseCode())));
verify(client,never()).inventory(any(SelfopRequestContext.class),eq("B"),anyInt(),anyInt());
assertTrue(capture.getAllValues().stream().anyMatch(s->"PENDING".equals(s.getValidationStatus())));
verify(egress).finishSample(1L,"VALIDATION_SAMPLE_SUFFICIENT");
verify(tasks,never()).updateById(any(SelfopSyncTask.class)); // quota root JSON cannot be overwritten by an earlier task snapshot
@@ -107,7 +109,7 @@ class SelfopValidationRunnerTest {
verify(client,never()).warehouses(any(SelfopRequestContext.class),anyInt(),anyInt());
verify(client,never()).detail(any(SelfopRequestContext.class),anyString(),anyString(),anyString());
verify(client,never()).inventory(any(SelfopRequestContext.class),eq("A"),anyInt(),anyInt());
verify(client).inventory(any(SelfopRequestContext.class),eq("B"),eq(1),eq(10));
verify(client,never()).inventory(any(SelfopRequestContext.class),eq("B"),anyInt(),anyInt());
assertEquals("SAMPLE_DONE",run.getStatus());
}
@Test void wrongPurposeCannotCallAnyClient() {
@@ -153,4 +155,47 @@ class SelfopValidationRunnerTest {
runner.run(task,run,lease);assertEquals("FAILED",run.getStatus());
verify(egress).failValidation(1L,"VALIDATION_STRUCTURE_REJECTED");verify(stages,never()).insert(any(SelfopSyncStage.class));
}
private void directoryOnly() {
config.getWarehouses().clear();task.setQueryScopeJson("{\"action\":\"WAREHOUSE_DISCOVERY\",\"sourceAccount\":\"demo\"}");
task.setValidationBudgetJson(SelfopValidationBudget.freezeWarehouseDiscovery(utcNow()));
}
@Test void warehouseDiscoveryReadsOnePageWithoutTimeWindowOrConfiguredWarehouseAndImmediatelyStops() {
directoryOnly();
var a=new LingxingAnalyticsClient.Warehouse("REAL-A","A",0,0,"America/New_York");
var b=new LingxingAnalyticsClient.Warehouse("REAL-B","B",0,0,"America/Los_Angeles");
when(client.warehouses(any(SelfopRequestContext.class),eq(1),eq(10))).thenReturn(page(1,2,1,List.of(a,b)));
runner.run(task,run,lease);
assertEquals("SAMPLE_DONE",run.getStatus());assertEquals("UNKNOWN",run.getCoverageStatus());assertNull(run.getPublishedAtUtc());
assertTrue(run.getObservedDimensionsJson().contains("REAL-A"));assertTrue(run.getObservedDimensionsJson().contains("REAL-B"));
verify(client,times(1)).warehouses(any(SelfopRequestContext.class),eq(1),eq(10));
verify(client,never()).discover(any(SelfopRequestContext.class),any(),any(),anyInt(),anyInt());
verify(client,never()).detail(any(SelfopRequestContext.class),anyString(),anyString(),anyString());
verify(client,never()).inventory(any(SelfopRequestContext.class),anyString(),anyInt(),anyInt());verifyNoInteractions(stages);
verify(egress).finishSample(1L,"VALIDATION_WAREHOUSE_DIRECTORY_COMPLETE");
}
@Test void truncatedDirectoryIsExplicitAndCannotRequestSecondPage() {
directoryOnly();when(client.warehouses(any(SelfopRequestContext.class),eq(1),eq(10)))
.thenReturn(page(1,11,2,List.of(new LingxingAnalyticsClient.Warehouse("A",null,0,0,null))));
runner.run(task,run,lease);verify(egress).finishSample(1L,"VALIDATION_WAREHOUSE_DIRECTORY_TRUNCATED");
verify(client,never()).warehouses(any(SelfopRequestContext.class),eq(2),anyInt());
}
@Test void directoryRateLimitTerminatesRootWithoutAnyBusinessRequest() {
directoryOnly();when(client.warehouses(any(SelfopRequestContext.class),eq(1),eq(10)))
.thenThrow(new AnalyticsClientException("LINGXING_WMS_RATE_LIMITED",true));
runner.run(task,run,lease);assertEquals("FAILED",run.getStatus());verify(egress).failValidation(1L,"LINGXING_WMS_RATE_LIMITED");
verify(client,never()).discover(any(SelfopRequestContext.class),any(),any(),anyInt(),anyInt());verifyNoInteractions(stages);
}
@Test void mixedAccountDirectoryOnlyFetchesDetailsOfFrozenWarehouse() {
var other=new LingxingAnalyticsClient.Summary("demo","customer","B","other","source",100,null,"2026-10-12 08:02:00");
when(client.discover(any(SelfopRequestContext.class),any(),any(),eq(1),eq(10))).thenReturn(page(1,2,1,List.of(other,summary("selected"))));
when(client.detail(any(SelfopRequestContext.class),eq("A"),eq("customer"),eq("selected")))
.thenReturn(new SelfopPayloads.Order("demo","customer","A","selected","source",100,null,"2026-10-12 08:02:00",List.of()));
runner.run(task,run,lease);assertEquals("SAMPLE_DONE",run.getStatus());
verify(client,never()).detail(any(SelfopRequestContext.class),eq("B"),anyString(),anyString());
verify(client,never()).inventory(any(SelfopRequestContext.class),eq("B"),anyInt(),anyInt());
}
@Test void missingOrMultipleFrozenWarehousesFailBeforeAnyProviderRequest() {
task.setQueryScopeJson("{\"action\":\"VALIDATION\",\"warehouses\":[\"A\",\"B\"]}");
runner.run(task,run,lease);assertEquals("FAILED",run.getStatus());verifyNoInteractions(client,stages);
}
}