From 2e2a354ba60635fa3bc8c59f2a41f535fe120ac3 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Wed, 14 Oct 2026 09:37:00 +0800 Subject: [PATCH] fix(selfop): retain turnover precision and scope reused publication runs --- .../mapper/SelfopCoordinationMapper.java | 15 ++- .../query/InventoryTurnoverPolicy.java | 6 +- .../analytics/query/InventoryTurnoverSql.java | 11 +- .../analytics/sync/SelfopSalesPublisher.java | 2 +- .../query/InventoryTurnoverPolicyTest.java | 38 +++++++ .../query/InventoryTurnoverSqlTest.java | 60 ++++++++++- .../sync/SelfopReusedRunScopeSqlTest.java | 102 ++++++++++++++++++ .../sync/SelfopSalesPublisherTest.java | 3 + 8 files changed, 227 insertions(+), 10 deletions(-) create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicyTest.java create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopReusedRunScopeSqlTest.java diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopCoordinationMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopCoordinationMapper.java index 2415972b9..b6465397d 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopCoordinationMapper.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopCoordinationMapper.java @@ -23,8 +23,19 @@ public interface SelfopCoordinationMapper { long barriers(@Param("account") String account, @Param("warehouse") String warehouse); @Update("UPDATE ll_selfop_sync_task SET status='SUCCEEDED',result_published_run_id=#{published},next_retry_at_utc=NULL,updated_at=#{now} WHERE scope_id=#{scope} AND scope_hash=#{hash} AND execution_purpose='BUSINESS' AND apply_state='APPLIED' AND (query_from_utc IS NULL OR (query_from_utc>=#{from} AND query_until_utc<=#{until})) AND status IN ('READY','RUNNING','RETRY_WAIT','DEAD')") int completeCoveredTasks(@Param("scope") long scope,@Param("hash") String hash,@Param("published") long published,@Param("from") LocalDateTime from,@Param("until") LocalDateTime until,@Param("now") LocalDateTime now); - @Update("UPDATE ll_selfop_sync_run r JOIN ll_selfop_sync_task t ON t.id=r.task_id SET r.status='REUSED',r.finished_at_utc=#{now},r.updated_at=#{now},r.progress_snapshot_json=COALESCE(r.progress_snapshot_json,CASE WHEN JSON_EXTRACT(t.progress_json,'$.attemptNo')=r.attempt_no THEN t.progress_json ELSE NULL END) WHERE t.result_published_run_id=#{published} AND r.id<>#{published} AND r.attempt_no=t.attempt_count AND r.execution_purpose='BUSINESS' AND r.status IN ('FETCHING','FACTS_APPLIED','BUILDING','READY')") - int completeReusedRuns(@Param("published") long published,@Param("now") LocalDateTime now); + @Update(""" + UPDATE ll_selfop_sync_run r JOIN ll_selfop_sync_task t ON t.id=r.task_id + SET r.status='REUSED',r.finished_at_utc=#{now},r.updated_at=#{now},r.progress_snapshot_json=COALESCE(r.progress_snapshot_json,CASE WHEN JSON_EXTRACT(t.progress_json,'$.attemptNo')=r.attempt_no THEN t.progress_json ELSE NULL END) + WHERE r.scope_id=#{scope} AND t.scope_id=#{scope} + AND r.source_account=#{account} AND t.source_account=#{account} + AND r.warehouse_code=#{warehouse} AND t.warehouse_code=#{warehouse} + AND r.scope_hash=#{hash} AND t.scope_hash=#{hash} + AND t.result_published_run_id=#{published} AND r.id<>#{published} + AND r.attempt_no=t.attempt_count AND r.execution_purpose='BUSINESS' AND t.execution_purpose='BUSINESS' + AND r.status IN ('FETCHING','FACTS_APPLIED','BUILDING','READY') + """) + int completeReusedRuns(@Param("scope") long scope,@Param("account") String account,@Param("warehouse") String warehouse, + @Param("hash") String hash,@Param("published") long published,@Param("now") LocalDateTime now); @Update("UPDATE ll_selfop_sync_task SET status='SUCCEEDED',result_published_run_id=#{published},next_retry_at_utc=NULL,updated_at=#{now} WHERE scope_id=#{scope} AND scope_hash=#{hash} AND execution_purpose='BUSINESS' AND task_type='SCHEDULED' AND apply_state='NOT_APPLICABLE' AND status='WAITING_SOURCE' AND planned_at_utc<=#{until}") int completeCoveredMarkers(@Param("scope") long scope,@Param("hash") String hash,@Param("published") long published,@Param("until") LocalDateTime until,@Param("now") LocalDateTime now); @Update("") diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicy.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicy.java index 505b6906d..b0a83256e 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicy.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicy.java @@ -49,11 +49,13 @@ public final class InventoryTurnoverPolicy { var r=normalized(raw);boolean owner="OWNER".equals(q.group()); Total physical=total(r,"physical",owner),available=total(r,"available",owner); var windows=windows(r,context.endDate().plusDays(1));var base=windows.stream().filter(w->w.days()==q.days()).findFirst().orElseThrow(); - var coverage=owner?null:scale(decimal(r,"coverage_days"));var average=scale(decimal(r,"average_physical")); + var coverage=owner?null:scale(decimal(r,"coverage_days"));var rawAverage=decimal(r,"average_physical");var average=scale(rawAverage); var turnover=scale(decimal(r,"turnover"));int sampleCount=count(r,"sample_count"); String coverageStatus=owner?"NOT_APPLICABLE":available.value()==null?"INVENTORY_UNKNOWN_OR_STALE":available.value().signum()==0?"NO_AVAILABLE_STOCK": base.outboundQuantity()==null||count(r,"unassessable")>0?"UNASSESSABLE":base.outboundQuantity().signum()==0?"NO_OUTBOUND_IN_WINDOW":"VALID"; - String sampleStatus=sampleCount missing=new ArrayList<>();String dates=text(r,"missing_dates"); if(dates!=null&&!dates.isBlank())for(String date:dates.split(","))missing.add(LocalDate.parse(date.substring(0,10))); List attention=new ArrayList<>(); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSql.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSql.java index cc7fdb683..12d76479b 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSql.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSql.java @@ -69,12 +69,12 @@ public final class InventoryTurnoverSql { .append(scope("")).append(" AND sales_run_id=#{p.salesRun} AND business_date0 AND quantity_unit='PCS' GROUP BY customer_code,sku_code)"); sql.append(", sku_base AS (SELECT c.customer_code,c.sku_code,c.product_name,c.evidence,c.unit_valid"); for(int n:List.of(7,15,30)) for(String suffix:List.of("outbound","observed","complete","partial")) sql.append(",w.w").append(n).append('_').append(suffix); - sql.append(",w.sample_count,w.average_physical,w.sample_mask,ls.last_outbound"); + sql.append(",w.sample_count,w.sample_total,w.average_physical,w.sample_mask,ls.last_outbound"); for(String field:List.of("physical","available","frozen","defective")) sql.append(",CASE WHEN #{p.stockFresh}=1 AND ip.quantity_unit='PCS' THEN st.").append(field).append(" END ").append(field); sql.append(" FROM profiles c JOIN sku_windows w ON w.customer_code=c.customer_code AND w.sku_code=c.sku_code LEFT JOIN last_sales ls ON ls.customer_code=c.customer_code AND ls.sku_code=c.sku_code LEFT JOIN stocks st ON st.snapshot_run_id=#{p.inventoryRun} AND st.customer_code=c.customer_code AND st.sku_code=c.sku_code ") .append("LEFT JOIN fixed_profiles ip ON ip.profile_run_id=#{p.inventoryRun} AND ip.customer_code=c.customer_code AND ip.sku_code=c.sku_code), sku_calculated AS (SELECT b.*,") .append("CASE WHEN available=0 THEN 0 WHEN available>0 AND w").append(p.query.days()).append("_outbound>0 AND evidence<=#{p.baseFromUtc} THEN available*#{p.query.days}/w").append(p.query.days()).append("_outbound END coverage_days,") - .append("CASE WHEN average_physical>0 AND evidence<=#{p.baseFromUtc} THEN w").append(p.query.days()).append("_outbound/average_physical END turnover,") + .append("CASE WHEN sample_total>0 AND evidence<=#{p.baseFromUtc} THEN w").append(p.query.days()).append("_outbound*#{p.query.days}/sample_total END turnover,") .append("CASE WHEN physical=0 THEN 0 WHEN physical>0 AND w30_outbound IS NOT NULL AND evidence<=#{p.fromUtc} THEN CASE WHEN w30_outbound=0 THEN 1 ELSE 0 END END no_outbound,") .append("CASE WHEN physical IS NULL OR available IS NULL OR w").append(p.query.days()).append("_outbound IS NULL OR evidence IS NULL OR evidence>#{p.baseFromUtc} THEN 1 ELSE 0 END unassessable FROM sku_base b), sku_result AS (") .append("SELECT b.*,CASE WHEN #{p.query.coverageDaysThreshold,jdbcType=INTEGER} IS NULL THEN NULL WHEN physical=0 OR available=0 THEN 0 WHEN coverage_days IS NOT NULL AND available>0 THEN CASE WHEN coverage_days>#{p.query.coverageDaysThreshold,jdbcType=INTEGER} THEN 1 ELSE 0 END WHEN available>0 AND w") @@ -115,8 +115,10 @@ public final class InventoryTurnoverSql { .append("JOIN fixed_profiles ip ON ip.profile_run_id=d.sample_run AND ip.customer_code=st.customer_code AND ip.sku_code=st.sku_code ") .append("WHERE ip.quantity_unit='PCS' AND st.physical IS NOT NULL), sku_sample_counts AS (SELECT customer_code,sku_code,") .append("COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) sample_count,") + // Keep the qualified sum exact; do not divide again by a rounded daily mean. + .append("CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) END sample_total,") .append("CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)/#{p.query.days} END average_physical,") - .append(sampleMask()).append(" sample_mask FROM sku_sample_days GROUP BY customer_code,sku_code), sku_window_counts AS (SELECT z.customer_code,z.sku_code,COALESCE(s.sample_count,0) sample_count,s.average_physical,COALESCE(s.sample_mask,0) sample_mask"); + .append(sampleMask()).append(" sample_mask FROM sku_sample_days GROUP BY customer_code,sku_code), sku_window_counts AS (SELECT z.customer_code,z.sku_code,COALESCE(s.sample_count,0) sample_count,s.sample_total,s.average_physical,COALESCE(s.sample_mask,0) sample_mask"); for(int n:List.of(7,15,30)) sql.append(",CASE WHEN z.w").append(n).append("_known+COALESCE(a.w").append(n).append("_known,0)>0 THEN COALESCE(a.w").append(n).append("_sum,0) END w").append(n).append("_observed") .append(",z.w").append(n).append("_complete+COALESCE(a.w").append(n).append("_complete,0) w").append(n).append("_complete") .append(",z.w").append(n).append("_partial+COALESCE(a.w").append(n).append("_partial,0) w").append(n).append("_partial"); @@ -167,10 +169,11 @@ public final class InventoryTurnoverSql { +"s.sample_physical FROM owner_candidates c CROSS JOIN calendar d LEFT JOIN owner_actual_days a ON a.customer_code=c.customer_code AND a.business_date=d.business_date " +"LEFT JOIN owner_sample_days s ON s.customer_code=c.customer_code AND s.business_date=d.business_date), owner_window_counts AS (SELECT customer_code" +windowAggregates()+",COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) sample_count," + +"CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) END sample_total," +"CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)/#{p.query.days} END average_physical," +sampleMask()+" sample_mask FROM owner_days GROUP BY customer_code), owner_windows AS (SELECT w.*"+formalWindows()+" FROM owner_window_counts w), owner_totals AS (" +totals("customer_code",true)+" FROM sku_result GROUP BY customer_code), owner_result AS (SELECT t.*,w.sample_count,w.average_physical,w.sample_mask," - +"CASE WHEN w.average_physical>0 THEN w.w"+p.query.days()+"_outbound/w.average_physical END turnover," + +"CASE WHEN w.sample_total>0 THEN w.w"+p.query.days()+"_outbound*#{p.query.days}/w.sample_total END turnover," +String.join(",",List.of(7,15,30).stream().flatMap(n->List.of("outbound","observed","complete","partial").stream().map(s->"w.w"+n+"_"+s)).toList()) +" FROM owner_totals t JOIN owner_windows w ON w.customer_code=t.customer_code) "; } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSalesPublisher.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSalesPublisher.java index 90f8ec677..9f6a656fd 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSalesPublisher.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSalesPublisher.java @@ -45,7 +45,7 @@ public class SelfopSalesPublisher { run.setStatus("PUBLISHED");run.setPublishedAtUtc(now);run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run); task.setResultPublishedRunId(runId);task.setStatus("SUCCEEDED");task.setUpdatedAt(now);tasks.updateById(task); locks.completeCoveredTasks(scopeId,scope.getScopeHash(),runId,run.getCoveredFromUtc(),run.getCoveredUntilUtc(),now); - locks.completeReusedRuns(runId,now); + locks.completeReusedRuns(scopeId,scope.getSourceAccount(),scope.getWarehouseCode(),scope.getScopeHash(),runId,now); locks.completeCoveredMarkers(scopeId,scope.getScopeHash(),runId,run.getCoveredUntilUtc(),now); } @Transactional diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicyTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicyTest.java new file mode 100644 index 000000000..ece0ea4c9 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicyTest.java @@ -0,0 +1,38 @@ +package com.baiyue.lastleg.selfoperated.analytics.query; + +import java.math.BigDecimal; +import java.util.*; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; +import static com.baiyue.lastleg.selfoperated.analytics.query.InventoryTurnoverResponses.*; + +class InventoryTurnoverPolicyTest { + static Context context() { + return new Context(null,"demo","W",InventoryTurnoverSqlTest.END,30,null,null,null,null,null,null,List.of(),List.of(),null); + } + private Row mapped(String average,String turnover,int samples) { + var raw=new HashMap();raw.put("average_physical",average);raw.put("turnover",turnover);raw.put("sample_count",samples); + raw.put("w30_outbound",BigDecimal.ONE);raw.put("w30_complete",30); + return InventoryTurnoverPolicy.row(raw,InventoryTurnoverSqlTest.request("SKU","ALL","TURNOVER_DESC",null,null,1,20),context()); + } + @Test void tinyPositiveAverageQualifiesBeforeSixPlaceOutputRounding() { + var row=mapped("0.0000000333","30000000",30); + assertEquals("VALID",row.turnoverStatus());assertTrue(row.sampleReasonCodes().isEmpty()); + assertEquals(new BigDecimal("0.000000"),row.averagePhysicalQuantity()); + assertEquals(new BigDecimal("30000000.000000"),row.quantityTurnoverCount()); + } + @Test void zeroUnknownAndIncompleteSamplesCannotLeakATurnoverValue() { + for(var input:List.of(new String[]{"0","ZERO_AVERAGE_STOCK"},new String[]{null,"INSUFFICIENT"})) { + var row=mapped(input[0],"30000000",30);assertNull(row.quantityTurnoverCount()); + assertEquals(input[1],row.turnoverStatus());assertEquals(List.of(input[1]),row.sampleReasonCodes()); + } + var row=mapped("1","30000000",29);assertNull(row.quantityTurnoverCount()); + assertEquals("INSUFFICIENT",row.turnoverStatus());assertEquals(List.of("INSUFFICIENT"),row.sampleReasonCodes()); + } + @Test void positiveAverageSeparatesMissingSalesFromKnownZeroSales() { + var missing=mapped("0.0000000333",null,30);assertNull(missing.quantityTurnoverCount()); + assertEquals("SALES_WINDOW_UNAVAILABLE",missing.turnoverStatus());assertEquals(List.of("SALES_WINDOW_UNAVAILABLE"),missing.sampleReasonCodes()); + var zero=mapped("0.0000000333","0",30);assertEquals("VALID",zero.turnoverStatus()); + assertEquals(new BigDecimal("0.000000"),zero.quantityTurnoverCount()); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSqlTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSqlTest.java index af7bed0fe..053b3bf6f 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSqlTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSqlTest.java @@ -362,12 +362,70 @@ class InventoryTurnoverSqlTest { assertEquals("SKU".equals(group)?0:3,report.stream().filter(r->"ROW".equals(r.get("section"))).count()); } } + @Test void tinyQualifiedDailySumUsesOneDivisionForSkuAndOwnerThroughProductionCtes() throws Exception { + seed("A","ONE",30);seed("A","TWO",30); + execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=CASE WHEN snapshot_run_id=100 AND product_quality=0 THEN CASE WHEN sku_code='ONE' THEN 0.000001 ELSE 0.000002 END ELSE 0 END WHERE snapshot_run_id>=100"); + execute("UPDATE ll_selfop_sku_daily_sales SET outbound_quantity=CASE WHEN business_date=DATE '"+END+"' THEN CASE WHEN sku_code='ONE' THEN 1 ELSE 2 END ELSE 0 END WHERE warehouse_code='W'"); + for(String group:List.of("SKU","OWNER")) { + var q=request(group,"ALL","TURNOVER_DESC",null,null,1,20);var p=input(q,30,30,true); + for(var row:reportRows(p)) { + assertTrue(new BigDecimal(row.get("average_physical").toString()).signum()>0); + decimal("30000000",row.get("turnover"));decimal("30",row.get("sample_count")); + var mapped=InventoryTurnoverPolicy.row(row,q,InventoryTurnoverPolicyTest.context()); + assertEquals("VALID",mapped.turnoverStatus());assertTrue(mapped.sampleReasonCodes().isEmpty()); + assertEquals(new BigDecimal("0.000000"),mapped.averagePhysicalQuantity()); + assertEquals(new BigDecimal("30000000.000000"),mapped.quantityTurnoverCount()); + } + for(var row:query("page",p))decimal("30000000",row.get("turnover")); + } + execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=0 WHERE snapshot_run_id>=100"); + for(String group:List.of("SKU","OWNER")) { + var q=request(group,"ALL","TURNOVER_DESC",null,null,1,20); + for(var row:reportRows(input(q,30,30,true))) { + decimal("0",row.get("average_physical"));assertNull(row.get("turnover")); + var mapped=InventoryTurnoverPolicy.row(row,q,InventoryTurnoverPolicyTest.context()); + assertEquals("ZERO_AVERAGE_STOCK",mapped.turnoverStatus());assertNull(mapped.quantityTurnoverCount()); + assertEquals(List.of("ZERO_AVERAGE_STOCK"),mapped.sampleReasonCodes()); + } + for(var row:reportRows(input(q,30,29,true))) { + assertNull(row.get("average_physical"));assertNull(row.get("turnover")); + var mapped=InventoryTurnoverPolicy.row(row,q,InventoryTurnoverPolicyTest.context()); + assertEquals("INSUFFICIENT",mapped.turnoverStatus());assertNull(mapped.quantityTurnoverCount()); + } + } + } + @Test void unroundedThresholdAndStableTiesSurviveActualReportPagination() throws Exception { + seed("OA","HIGH",30);seed("OA","A",30);seed("OZ","A",30);seed("OA","LOW",30); + execute("UPDATE ll_selfop_inventory_snapshot SET available_quantity=CASE sku_code WHEN 'HIGH' THEN 200.000001 WHEN 'LOW' THEN 199.999999 ELSE 200 END WHERE snapshot_run_id=22 AND product_quality=0"); + var gt=request("SKU","COVERAGE_GT","COVERAGE_DESC",null,20,1,20); + var greater=reportRows(input(gt,30,30,true));assertEquals(List.of("HIGH"),greater.stream().map(r->r.get("sku_code")).toList()); + assertEquals(new BigDecimal("20.000000"),InventoryTurnoverPolicy.row(greater.get(0),gt,InventoryTurnoverPolicyTest.context()).inventoryCoverageDays()); + var le=request("SKU","COVERAGE_LE","COVERAGE_DESC",null,20,1,20); + assertEquals(List.of("A/OA","A/OZ","LOW/OA"),reportRows(input(le,30,30,true)).stream().map(r->r.get("sku_code")+"/"+r.get("customer_code")).toList()); + var expected=List.of("HIGH/OA","A/OA","A/OZ","LOW/OA"); + for(int page=1;page<=4;page++) { + var q=request("SKU","ALL","COVERAGE_DESC",null,20,page,1);var result=reportRows(input(q,30,30,true)); + assertEquals(1,result.size());assertEquals(expected.get(page-1),result.get(0).get("sku_code")+"/"+result.get(0).get("customer_code")); + } + } + @Test void turnoverOrderingUsesUnroundedRatioBeforeStableSkuAndOwnerPages() throws Exception { + seed("Z","HIGH",30);seed("A","TIE",30);seed("B","TIE",30);seed("L","LOW",30); + execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=CASE WHEN product_quality=0 THEN 10 ELSE 0 END WHERE snapshot_run_id>=100"); + execute("UPDATE ll_selfop_sku_daily_sales SET outbound_quantity=CASE WHEN business_date=DATE '"+END+"' THEN CASE sku_code WHEN 'HIGH' THEN 10.000001 WHEN 'LOW' THEN 9.999999 ELSE 10 END ELSE 0 END WHERE warehouse_code='W'"); + for(String group:List.of("SKU","OWNER")) for(int page=1;page<=4;page++) { + var q=request(group,"ALL","TURNOVER_DESC",null,null,page,1); + var result=reportRows(input(q,30,30,true));assertEquals(1,result.size()); + assertEquals(List.of("Z","A","B","L").get(page-1),result.get(0).get("customer_code")); + var mapped=InventoryTurnoverPolicy.row(result.get(0),q,InventoryTurnoverPolicyTest.context()); + assertEquals("VALID",mapped.turnoverStatus());assertEquals(new BigDecimal("1.000000"),mapped.quantityTurnoverCount()); + } + } private static com.fasterxml.jackson.databind.JsonNode payload(Map row) throws Exception { var json=new com.fasterxml.jackson.databind.ObjectMapper().enable(com.fasterxml.jackson.databind.DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);Object raw=row.get("payload"); return raw instanceof byte[] bytes?json.readTree(bytes):json.readTree(raw.toString()); } private List> reportRows(InventoryTurnoverSql.Input p) throws Exception { - var result=new ArrayList>();var json=new com.fasterxml.jackson.databind.ObjectMapper(); + var result=new ArrayList>();var json=new com.fasterxml.jackson.databind.ObjectMapper().enable(com.fasterxml.jackson.databind.DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS); for(var row:query("report",p)) if("ROW".equals(row.get("section"))) result.add(json.convertValue(payload(row),new com.fasterxml.jackson.core.type.TypeReference>(){})); return result; diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopReusedRunScopeSqlTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopReusedRunScopeSqlTest.java new file mode 100644 index 000000000..e4c6f3946 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopReusedRunScopeSqlTest.java @@ -0,0 +1,102 @@ +package com.baiyue.lastleg.selfoperated.analytics.sync; + +import com.baiyue.lastleg.selfoperated.analytics.mapper.SelfopCoordinationMapper; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.ibatis.mapping.BoundSql; +import org.apache.ibatis.mapping.MappedStatement; +import org.apache.ibatis.scripting.defaults.DefaultParameterHandler; +import org.apache.ibatis.session.Configuration; +import org.junit.jupiter.api.*; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import java.sql.*; +import java.time.LocalDateTime; +import java.util.*; +import static org.junit.jupiter.api.Assertions.*; + +/** Executes production JOIN/predicates/assignments offline. Only MySQL UPDATE JOIN syntax is + * adapted to H2 MERGE; this does not prove MySQL optimizer choices or lock isolation. */ +public class SelfopReusedRunScopeSqlTest { + Connection db; MappedStatement statement; BoundSql bound; + final Map params=Map.of("scope",7L,"account","SYNTH-ACCOUNT","warehouse","SYNTH-A", + "hash","scope-hash","published",900L,"now",LocalDateTime.of(2026,10,14,0,0)); + + @BeforeEach void setup() throws Exception { + db=DriverManager.getConnection("jdbc:h2:mem:"+UUID.randomUUID()+";MODE=MySQL;DATABASE_TO_LOWER=TRUE","sa",""); + execute("CREATE TABLE ll_selfop_sync_task(id BIGINT PRIMARY KEY,scope_id BIGINT,source_account VARCHAR,warehouse_code VARCHAR,scope_hash VARCHAR,result_published_run_id BIGINT,attempt_count INT,execution_purpose VARCHAR,progress_json VARCHAR)"); + execute("CREATE TABLE ll_selfop_sync_run(id BIGINT PRIMARY KEY,task_id BIGINT,scope_id BIGINT,source_account VARCHAR,warehouse_code VARCHAR,scope_hash VARCHAR,attempt_no INT,execution_purpose VARCHAR,status VARCHAR,finished_at_utc TIMESTAMP,updated_at TIMESTAMP,progress_snapshot_json VARCHAR)"); + execute("CREATE ALIAS JSON_EXTRACT FOR '"+getClass().getName()+".jsonExtract'"); + var config=new Configuration();config.addMapper(SelfopCoordinationMapper.class); + statement=config.getMappedStatement(SelfopCoordinationMapper.class.getName()+".completeReusedRuns"); + bound=statement.getBoundSql(params); + } + @AfterEach void close() throws Exception {db.close();} + public static String jsonExtract(String body,String path) throws Exception { + if(body==null)return null;assertEquals("$.attemptNo",path); + var value=new ObjectMapper().readTree(body).get("attemptNo");return value==null?null:value.asText(); + } + void execute(String sql,Object...args) throws Exception { + try(var p=db.prepareStatement(sql)){for(int n=0;n row(long id) throws Exception { + try(var p=db.prepareStatement("SELECT * FROM ll_selfop_sync_run WHERE id=?")){p.setLong(1,id);try(var r=p.executeQuery()){assertTrue(r.next());var out=new TreeMap();for(int n=1;n<=r.getMetaData().getColumnCount();n++)out.put(r.getMetaData().getColumnLabel(n),r.getObject(n));return out;}} + } + int update() throws Exception { + String sql=bound.getSql().replaceAll("\\s+"," ").trim(); + int set=sql.indexOf(" SET "),where=sql.indexOf(" WHERE ");assertTrue(sql.startsWith("UPDATE ")&&set>0&&where>set); + String from=sql.substring(7,set),predicate=sql.substring(where+7); + // Assignment order preserves MyBatis parameter order. No test-owned selection predicate. + var projection=new ArrayList();var assignments=new ArrayList(); + for(String assignment:sql.substring(set+5,where).split(",(?=r\\.)")){ + int equal=assignment.indexOf('=');String column=assignment.substring(0,equal).replace("r.",""); + String expression=assignment.substring(equal+1); + // H2 cannot infer a bare parameter's type in a derived table; production columns are timestamps. + if(expression.equals("?")){assertTrue(column.equals("finished_at_utc")||column.equals("updated_at"));expression="CAST(? AS TIMESTAMP)";} + projection.add(expression+" AS "+column);assignments.add("destination."+column+"=source."+column); + } + String adapted="MERGE INTO ll_selfop_sync_run destination USING (SELECT r.id,"+String.join(",",projection)+" FROM "+from+" WHERE "+predicate+") source ON destination.id=source.id WHEN MATCHED THEN UPDATE SET "+String.join(",",assignments); + try(var p=db.prepareStatement(adapted)){new DefaultParameterHandler(statement,params,bound).setParameters(p);return p.executeUpdate();} + } + @Test void nonemptyMatchingAttemptsCompleteAllFourEligibleStatusesAndKeepProgress() throws Exception { + String[] statuses={"FETCHING","FACTS_APPLIED","BUILDING","READY"}; + for(int n=0;n execute("UPDATE ll_selfop_sync_run SET attempt_no=1 WHERE id=12"); + case "wrong-result" -> execute("UPDATE ll_selfop_sync_task SET result_published_run_id=901 WHERE id=1012"); + case "run-validation" -> execute("UPDATE ll_selfop_sync_run SET execution_purpose='VALIDATION' WHERE id=12"); + case "task-validation" -> execute("UPDATE ll_selfop_sync_task SET execution_purpose='VALIDATION' WHERE id=1012"); + case "null-scope" -> execute("UPDATE ll_selfop_sync_task SET scope_id=NULL WHERE id=1012"); + default -> execute("UPDATE ll_selfop_sync_run SET status=? WHERE id=12",fault); + } + var before=row(12);assertEquals(1,update());assertEquals("REUSED",row(11).get("status"));assertEquals(before,row(12)); + } + @Test void mismatchedProgressAttemptIsNotCopiedIntoCompletedRun() throws Exception { + pair(11,"READY");execute("UPDATE ll_selfop_sync_task SET progress_json=? WHERE id=1011","{\"attemptNo\":1}"); + assertEquals(1,update());assertEquals("REUSED",row(11).get("status"));assertNull(row(11).get("progress_snapshot_json")); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSalesPublisherTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSalesPublisherTest.java index 9364a17bc..35b370ead 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSalesPublisherTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSalesPublisherTest.java @@ -122,6 +122,9 @@ class SelfopSalesPublisherTest { ordered.verify(runs).update(isNull(),any()); ordered.verify(scopes).updateById(any(SelfopSyncScope.class)); ordered.verify(runs).updateById(any(SelfopSyncRun.class)); ordered.verify(tasks).updateById(any(SelfopSyncTask.class)); + ordered.verify(locks).completeCoveredTasks(eq(1L),eq(scope.getScopeHash()),eq(10L),eq(run.getCoveredFromUtc()),eq(run.getCoveredUntilUtc()),any()); + ordered.verify(locks).completeReusedRuns(eq(1L),eq(scope.getSourceAccount()),eq(scope.getWarehouseCode()),eq(scope.getScopeHash()),eq(10L),any()); + ordered.verify(locks).completeCoveredMarkers(eq(1L),eq(scope.getScopeHash()),eq(10L),eq(run.getCoveredUntilUtc()),any()); assertNotNull(SelfopSalesPublisher.class.getMethod("publish",long.class,long.class,String.class,long.class).getAnnotation(Transactional.class)); assertEquals(10L,task.getResultPublishedRunId());assertNotNull(run.getProgressSnapshotJson()); // Publisher constructor has no inventory repositories or inventory publication prerequisite.