diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopCollectionPolicy.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopCollectionPolicy.java index 249a563ae..90c4e6f2c 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopCollectionPolicy.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopCollectionPolicy.java @@ -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; + } } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsInventorySql.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsInventorySql.java index e1628dd88..fe092e308 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsInventorySql.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsInventorySql.java @@ -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 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 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 args){var p=(AnalyticsSqlRequest)args.get("p");StringBuilder sql=new StringBuilder(cte(p)).append("SELECT COUNT(*) sku_count"); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryService.java index abceab4fd..b9199c9cb 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryService.java @@ -132,10 +132,11 @@ public class AnalyticsQueryService { } private AnalyticsResponses.Page 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 statistics=mapper.rankingStats(p); + long total=statistics.isEmpty()?0:statistics.get(0).total(); List rows=q.offset()>=total?List.of():mapper.page(p).stream().map(r->metric(s,r)).toList(); Map 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){ diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadMapper.java index b96b79445..0fa4d19a4 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadMapper.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadMapper.java @@ -20,9 +20,9 @@ public interface AnalyticsReadMapper { List 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 excluded(@Param("p") AnalyticsSqlRequest request); + record RankingStat(Long total,String reason,Long excludedCount){} + @SelectProvider(type=AnalyticsReadSql.class,method="rankingStats") + List 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}") diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadSql.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadSql.java index beefeefef..4599fa82e 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadSql.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadSql.java @@ -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'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 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 parameters){ + public static String rankingStats(java.util.Map 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} "; diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressService.java index 9bd106d97..831136856 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressService.java @@ -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"); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopValidationBudget.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopValidationBudget.java index 617db45e4..ad63c290a 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopValidationBudget.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopValidationBudget.java @@ -11,6 +11,7 @@ public final class SelfopValidationBudget { public Set detailKeys=new LinkedHashSet<>(); public Map 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); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopCommandService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopCommandService.java index 87a5540f5..d8a9b76f0 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopCommandService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopCommandService.java @@ -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 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","任务已成功完成,无需重试"), diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlanner.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlanner.java index 378e878e5..3752cbd17 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlanner.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlanner.java @@ -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 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); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java index e33da0bce..9ea8ac8cf 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java @@ -30,14 +30,7 @@ public class SelfopTaskService { public List initialize() { identity(config.getSourceAccount(),64); require(!config.getWarehouses().isEmpty(),"SELFOP_WAREHOUSE_UNCONFIGURED"); var now=utcNow(); - var a=accounts.selectOne(new QueryWrapper().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().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().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().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().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() .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()); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopValidationRunner.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopValidationRunner.java index 3ad5c8a3a..2fc728238 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopValidationRunner.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopValidationRunner.java @@ -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 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.detailIndexclient.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; diff --git a/lastleg-service/src/main/resources/openapi/self-operated-analytics-sync-v1.yaml b/lastleg-service/src/main/resources/openapi/self-operated-analytics-sync-v1.yaml index b9a0b893c..f8792fd24 100644 --- a/lastleg-service/src/main/resources/openapi/self-operated-analytics-sync-v1.yaml +++ b/lastleg-service/src/main/resources/openapi/self-operated-analytics-sync-v1.yaml @@ -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' } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryServiceTest.java index 0310630cc..5dbdf443d 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryServiceTest.java @@ -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()); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadMapperTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadMapperTest.java index c4c2a38a3..b74055ebc 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadMapperTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsReadMapperTest.java @@ -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 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=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")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 rankingStats(AnalyticsSqlRequest p){ + return (List)(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 inventory(String attention,Integer threshold,String sort,long offset,int size){ diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java index 418b13912..efad1e3c9 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java @@ -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");} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopCommandServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopCommandServiceTest.java index d749e84d4..3d085a894 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopCommandServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopCommandServiceTest.java @@ -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"); } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java index 39d8eefe2..9acb858ec 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java @@ -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"); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java index cca66f8c9..79fbbda73 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java @@ -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"); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopValidationRunnerTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopValidationRunnerTest.java index 5c44540a6..71b66aeef 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopValidationRunnerTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopValidationRunnerTest.java @@ -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); + } }