java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(selfop): publish proven manual sales windows for configured warehouses

Browse Source
This commit is contained in:
ouyangxq
2026-10-13 00:30:48 +08:00
parent ae1a809357
commit a84a9b730e
6 changed files with 63 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapper.java
+11 -2
View File
@@ -115,7 +115,7 @@ public interface OutboundAggregateMapper {
@Select("""
<script>
SELECT id,query_from_utc,query_until_utc FROM ll_selfop_sync_task WHERE scope_id=#{scopeId} AND scope_hash=#{scopeHash}
AND execution_purpose='BUSINESS' AND apply_state='APPLIED'
AND execution_purpose='BUSINESS' AND validation_root_task_id IS NULL AND task_role='WAREHOUSE_SYNC' AND apply_state='APPLIED'
AND query_from_utc IS NOT NULL AND query_until_utc IS NOT NULL
AND JSON_UNQUOTE(JSON_EXTRACT(routing_manifest_json,'$.directoryComplete'))='true'
<if test='id != null'>AND (query_from_utc > #{from} OR (query_from_utc=#{from} AND id > #{id}))</if>
@@ -125,6 +125,15 @@ public interface OutboundAggregateMapper {
List<SelfopSyncTask> coverageBatch(@Param("scopeId") Long scopeId,@Param("scopeHash") String scopeHash,
@Param("from") LocalDateTime from,@Param("id") Long id);
@Select("""
SELECT MIN(query_from_utc) FROM ll_selfop_sync_task
WHERE scope_id=#{scopeId} AND scope_hash=#{scopeHash}
AND execution_purpose='BUSINESS' AND validation_root_task_id IS NULL AND task_role='WAREHOUSE_SYNC' AND apply_state='APPLIED'
AND query_from_utc IS NOT NULL AND query_until_utc > query_from_utc
AND JSON_UNQUOTE(JSON_EXTRACT(routing_manifest_json,'$.directoryComplete'))='true'
""")
LocalDateTime firstProvenCoverageFrom(@Param("scopeId") Long scopeId,@Param("scopeHash") String scopeHash);
@Select("""
<script>
SELECT CAST(#{initial} AS DATETIME) AS covered_from_utc,MAX(until_at) AS covered_until_utc
@@ -136,7 +145,7 @@ public interface OutboundAggregateMapper {
OVER (ORDER BY from_at,until_at ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS previous_until
FROM (
SELECT query_from_utc AS from_at,query_until_utc AS until_at FROM ll_selfop_sync_task
WHERE scope_id=#{scopeId} AND scope_hash=#{scopeHash} AND execution_purpose='BUSINESS' AND apply_state='APPLIED'
WHERE scope_id=#{scopeId} AND scope_hash=#{scopeHash} AND execution_purpose='BUSINESS' AND validation_root_task_id IS NULL AND task_role='WAREHOUSE_SYNC' AND apply_state='APPLIED'
AND query_from_utc IS NOT NULL AND query_until_utc > query_from_utc AND query_until_utc >= #{initial}
AND JSON_UNQUOTE(JSON_EXTRACT(routing_manifest_json,'$.directoryComplete'))='true'
UNION ALL SELECT CAST(#{initial} AS DATETIME),CAST(#{initial} AS DATETIME)
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateTransaction.java
+6 -2
View File
@@ -136,8 +136,12 @@ public class OutboundCandidateTransaction {
SelfopSupport.require(aggregates.invalidFacts(scope.getSourceAccount(),scope.getWarehouseCode(),scope.getSourceTimezone(),scope.getBusinessTimezone(),unit)==0,"FACT_TIME_UNIT_OR_EVIDENCE_INVALID");
}
private OutboundCandidateBuilder.Coverage coverage(SelfopSyncScope scope) {
if(scope.getInitialFromUtc()==null)return new OutboundCandidateBuilder.Coverage(null,null);
SelfopSyncRun result=aggregates.contiguousCoverage(scope.getId(),scope.getScopeHash(),scope.getInitialFromUtc(),scope.getDiscoveryFromUtc(),scope.getDiscoveryUntilUtc());
// A manual first slice has no automatic backfill anchor. Only accepted, fully applied
// coverage may establish its start; an existing planned anchor must never move past gaps.
LocalDateTime initial=scope.getInitialFromUtc();
if(initial==null)initial=aggregates.firstProvenCoverageFrom(scope.getId(),scope.getScopeHash());
if(initial==null)return new OutboundCandidateBuilder.Coverage(null,null);
SelfopSyncRun result=aggregates.contiguousCoverage(scope.getId(),scope.getScopeHash(),initial,scope.getDiscoveryFromUtc(),scope.getDiscoveryUntilUtc());
return result==null?new OutboundCandidateBuilder.Coverage(null,null):new OutboundCandidateBuilder.Coverage(result.getCoveredFromUtc(),result.getCoveredUntilUtc());
}
private Checkpoint checkpoint(SelfopSyncTask task) {
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java
+1
View File
@@ -208,6 +208,7 @@ public class SelfopDiscoveryService {
if(read(parent.getQueryScopeJson(),Query.class).parentTaskId()!=null)return;
tx.executeWithoutResult(s->{leases.verify(lease);if(!manifest.isDirectoryComplete())return;
for(var scope:scopes.selectList(new QueryWrapper<SelfopSyncScope>().eq("source_account",parent.getSourceAccount()).eq("data_family","OUTBOUND"))){
if(!config.getWarehouses().containsKey(scope.getWarehouseCode()))continue;
var own=new SelfopRoutingManifest();own.setDirectoryComplete(true);own.setTotal(0L);own.setPages(manifest.getPages());
for(var e:manifest.getObjects().entrySet()){
var ref=e.getValue();var previous=orders.selectOne(new QueryWrapper<SelfopOutboundOrder>().eq("source_account",parent.getSourceAccount()).eq("customer_code",ref.getCustomerCode()).eq("delivery_no",ref.getDeliveryNo()));
lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapperTest.java
+10 -1
View File
@@ -20,7 +20,7 @@ public class OutboundAggregateMapperTest {
sql("CREATE TABLE ll_selfop_outbound_item (id BIGINT PRIMARY KEY,order_id BIGINT,sku_code VARCHAR(128),product_name VARCHAR(255),quantity_unit VARCHAR(32),actual_quantity DECIMAL(20,6))");
sql("CREATE TABLE ll_selfop_sync_stage (id BIGINT PRIMARY KEY,run_id BIGINT,warehouse_code VARCHAR(64),object_type VARCHAR(32),validation_status VARCHAR(24),observation_sequence BIGINT,observed_at_utc TIMESTAMP)");
sql("CREATE TABLE ll_selfop_sync_run (id BIGINT PRIMARY KEY,source_account VARCHAR(64),task_id BIGINT,execution_purpose VARCHAR(24))");
sql("CREATE TABLE ll_selfop_sync_task (id BIGINT PRIMARY KEY,execution_purpose VARCHAR(24),validation_root_task_id BIGINT,scope_id BIGINT,scope_hash VARCHAR(64),apply_state VARCHAR(24),query_from_utc TIMESTAMP,query_until_utc TIMESTAMP,routing_manifest_json VARCHAR)");
sql("CREATE TABLE ll_selfop_sync_task (id BIGINT PRIMARY KEY,execution_purpose VARCHAR(24),validation_root_task_id BIGINT,task_role VARCHAR(32) DEFAULT 'WAREHOUSE_SYNC',scope_id BIGINT,scope_hash VARCHAR(64),apply_state VARCHAR(24),query_from_utc TIMESTAMP,query_until_utc TIMESTAMP,routing_manifest_json VARCHAR)");
sql("CREATE ALIAS JSON_EXTRACT FOR 'com.baiyue.lastleg.selfoperated.analytics.outbound.OutboundAggregateMapperTest.jsonExtract'");
sql("CREATE ALIAS JSON_UNQUOTE FOR 'com.baiyue.lastleg.selfoperated.analytics.outbound.OutboundAggregateMapperTest.jsonUnquote'");
sql("CREATE TABLE ll_selfop_sku_daily_sales (id BIGINT PRIMARY KEY,created_at TIMESTAMP,updated_at TIMESTAMP,sales_run_id BIGINT,source_account VARCHAR(64),customer_code VARCHAR(64),warehouse_code VARCHAR(64),sku_code VARCHAR(128),business_date DATE,outbound_quantity DECIMAL(20,6),outbound_order_count BIGINT,quantity_unit VARCHAR(32))");
@@ -80,6 +80,15 @@ public class OutboundAggregateMapperTest {
return new com.fasterxml.jackson.databind.ObjectMapper().readTree(value).path("directoryComplete").asText();
}
public static String jsonUnquote(String value) {return value;}
@Test void firstManualCoverageRequiresMatchingScopeAppliedCompleteBusinessEvidence() throws Exception {
sql("INSERT INTO ll_selfop_sync_task (id,execution_purpose,scope_id,scope_hash,apply_state,query_from_utc,query_until_utc,routing_manifest_json) VALUES (1,'VALIDATION',1,'h','APPLIED','2026-10-01 00:00:00','2026-10-01 00:05:00','{\"directoryComplete\":true}'),(2,'BUSINESS',1,'h','NOT_STARTED','2026-10-02 00:00:00','2026-10-02 00:05:00','{\"directoryComplete\":true}'),(3,'BUSINESS',1,'h','APPLIED','2026-10-03 00:00:00','2026-10-03 00:05:00','{\"directoryComplete\":false}'),(4,'BUSINESS',2,'h','APPLIED','2026-10-04 00:00:00','2026-10-04 00:05:00','{\"directoryComplete\":true}'),(5,'BUSINESS',1,'old','APPLIED','2026-10-05 00:00:00','2026-10-05 00:05:00','{\"directoryComplete\":true}')");
sql("INSERT INTO ll_selfop_sync_task (id,execution_purpose,scope_id,scope_hash,apply_state,query_from_utc,query_until_utc,routing_manifest_json,task_role,validation_root_task_id) VALUES (7,'BUSINESS',1,'h','APPLIED','2026-10-01 00:00:00','2026-10-01 00:05:00','{\"directoryComplete\":true}','ACCOUNT_DISCOVERY',NULL),(8,'BUSINESS',1,'h','APPLIED','2026-10-01 00:00:00','2026-10-01 00:05:00','{\"directoryComplete\":true}','WAREHOUSE_SYNC',99)");
assertNull(mapper.firstProvenCoverageFrom(1L,"h"));
sql("INSERT INTO ll_selfop_sync_task (id,execution_purpose,scope_id,scope_hash,apply_state,query_from_utc,query_until_utc,routing_manifest_json) VALUES (6,'BUSINESS',1,'h','APPLIED','2026-10-10 01:00:00','2026-10-10 01:05:00','{\"directoryComplete\":true}')");
var start=mapper.firstProvenCoverageFrom(1L,"h");assertEquals(LocalDateTime.of(2026,10,10,1,0),start);
assertEquals(start.plusMinutes(5),mapper.contiguousCoverage(1L,"h",start,null,null).getCoveredUntilUtc());
}
@Test void databaseCoverageJoinsOverlapsAndStopsAtFirstUnprovedGap() throws Exception {
sql("INSERT INTO ll_selfop_sync_task (id,execution_purpose,scope_id,scope_hash,apply_state,query_from_utc,query_until_utc,routing_manifest_json) VALUES (1,'BUSINESS',1,'h','APPLIED','2026-10-01 00:00:00','2026-10-03 00:00:00','{\"directoryComplete\":true}'),(2,'BUSINESS',1,'h','APPLIED','2026-10-02 00:00:00','2026-10-04 00:00:00','{\"directoryComplete\":true}'),(3,'BUSINESS',1,'h','APPLIED','2026-10-05 00:00:00','2026-10-06 00:00:00','{\"directoryComplete\":true}'),(4,'VALIDATION',1,'h','APPLIED','2026-10-04 00:00:00','2026-10-05 00:00:00','{\"directoryComplete\":true}')");
var coverage=mapper.contiguousCoverage(1L,"h",LocalDateTime.of(2026,10,1,0,0),null,null);
lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilderTest.java
+23
View File
@@ -38,6 +38,29 @@ class OutboundCandidateBuilderTest {
SelfopSyncRun coverage=new SelfopSyncRun();coverage.setCoveredFromUtc(task.getQueryFromUtc());coverage.setCoveredUntilUtc(task.getQueryUntilUtc());
when(aggregates.contiguousCoverage(1L,"scope",scope.getInitialFromUtc(),null,null)).thenReturn(coverage);
}
@Test void manualFirstSliceUsesOnlyProvenIntervalWithoutInventingFullDays() {
scope.setInitialFromUtc(null);
var start=LocalDateTime.of(2026,10,9,17,0); var end=start.plusMinutes(5);
when(aggregates.firstProvenCoverageFrom(1L,"scope")).thenReturn(start);
var coverage=new SelfopSyncRun();coverage.setCoveredFromUtc(start);coverage.setCoveredUntilUtc(end);
when(aggregates.contiguousCoverage(1L,"scope",start,null,null)).thenReturn(coverage);
builder.build(1L,10L,"worker",2);
assertEquals(start,run.getCoveredFromUtc());assertEquals(end,run.getCoveredUntilUtc());
assertEquals("COMPLETE",run.getCoverageStatus());assertNull(scope.getInitialFromUtc());
assertTrue(run.getCompleteDayFrom()==null || !run.getCompleteDayFrom().isBefore(run.getCompleteDayUntil()));
}
@Test void missingProvenSliceCannotInventCoverage() {
scope.setInitialFromUtc(null);builder.build(1L,10L,"worker",2);
assertEquals("UNKNOWN",run.getCoverageStatus());assertNull(run.getCoveredFromUtc());assertNull(run.getCoveredUntilUtc());
verify(aggregates,never()).contiguousCoverage(any(),any(),any(),any(),any());
}
@Test void existingPlannedAnchorCannotMoveForwardPastMissingHistory() {
when(aggregates.contiguousCoverage(1L,"scope",scope.getInitialFromUtc(),null,null)).thenReturn(null);
builder.build(1L,10L,"worker",2);
assertEquals("UNKNOWN",run.getCoverageStatus());assertNull(run.getCoveredUntilUtc());
verify(aggregates,never()).firstProvenCoverageFrom(any(),any());
}
@Test void fullRebuildWithNoCurrentFactsRetractsAllRemovedEvidence() {
builder.build(1L,10L,"worker",2);
verify(profiles).delete(any()); verify(aggregates,never()).insertProfiles(any());
lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java
+12
View File
@@ -54,6 +54,7 @@ class SelfopDiscoveryServiceTest {
scope = new SelfopSyncScope(); scope.setId(10L); scope.setSourceAccount("demo"); scope.setWarehouseCode("A");
scope.setDataFamily("OUTBOUND"); scope.setScopeHash("warehouse-hash"); scope.setScopeJson("{}"); scope.setSourceTimezone("UTC");
config.setCollectionMode("BUSINESS"); config.setEnvironment("production"); config.setPageSize(2); config.setSourceQueryTimezone("UTC"); config.setQuantityUnit("PCS");
config.getWarehouses().put("A", new SelfopAnalyticsProperties.Warehouse());
when(scopes.selectList(any())).thenReturn(List.of(scope));
when(leases.nextObservation(lease)).thenReturn(101L, 102L, 103L);
when(taskService.create(any(SelfopSyncTask.class))).thenAnswer(inv -> {
@@ -89,6 +90,17 @@ class SelfopDiscoveryServiceTest {
summary("one", "2026-10-11 12:00:00"), summary("two", "2026-10-11 12:00:00")));
}
@Test void removedWarehouseScopeDoesNotReceiveRoutedTask() {
var removed = new SelfopSyncScope(); removed.setId(11L); removed.setSourceAccount("demo");
removed.setWarehouseCode("REMOVED"); removed.setDataFamily("OUTBOUND");
removed.setScopeHash("old-hash"); removed.setScopeJson("{}"); removed.setSourceTimezone("UTC");
when(scopes.selectList(any())).thenReturn(List.of(scope, removed));
directoryTwoOrders(); service.collect(task, run, lease);
assertEquals(1, children.size());
assertTrue(children.values().stream().allMatch(child -> "A".equals(child.getWarehouseCode())));
assertEquals("ROUTED", run.getStatus());
}
@Test void paginatedLocateYieldsDurableCursorThenResumesWithoutFiniteFailure() {
config.setPageSize(1); account.setRequestBudget(1);
task.setQueryScopeJson(json(new SelfopDiscoveryService.Query(null, null, "target", "customer", null)));