From c640fa2d0d4d9a926f272754679101c23645d6c4 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Wed, 14 Oct 2026 08:15:16 +0800 Subject: [PATCH] feat(selfop): prefer list SKU quantities with targeted detail fallback --- .../dto/lingxing/LingxingWmsDtos.java | 3 +- .../provider/lingxing/LingxingWmsClient.java | 13 ++- .../resources/openapi/lingxing-wms-v1.yaml | 10 +++ .../LingxingInternalControllerTest.java | 2 +- .../lingxing/LingxingWmsClientTest.java | 11 +++ .../client/LingxingAnalyticsClient.java | 12 ++- .../client/RestLingxingAnalyticsClient.java | 13 ++- .../model/SelfopRoutingManifest.java | 1 + .../analytics/sync/SelfopBudget.java | 3 +- .../sync/SelfopDiscoveryService.java | 75 ++++++++++++----- .../sync/SelfopValidationRunner.java | 6 +- .../RestLingxingAnalyticsClientTest.java | 15 ++++ .../sync/SelfopDiscoveryServiceTest.java | 83 ++++++++++++++++++- .../sync/SelfopValidationRunnerTest.java | 4 +- 14 files changed, 220 insertions(+), 31 deletions(-) diff --git a/integration-service/src/main/java/com/baiyue/integration/dto/lingxing/LingxingWmsDtos.java b/integration-service/src/main/java/com/baiyue/integration/dto/lingxing/LingxingWmsDtos.java index 52257c682..1e43243bb 100644 --- a/integration-service/src/main/java/com/baiyue/integration/dto/lingxing/LingxingWmsDtos.java +++ b/integration-service/src/main/java/com/baiyue/integration/dto/lingxing/LingxingWmsDtos.java @@ -45,8 +45,9 @@ public final class LingxingWmsDtos { } // Preserve source strings; the collector interprets confirmed Asia/Shanghai source time. + public record ListSku(String productSku, String productName, BigDecimal qty) { } public record OrderSummary(String deliveryNo, String sourceNo, String customerCode, String whCode, - Integer status, String createTime, String outboundTime) { } + Integer status, String createTime, String outboundTime, List productList) { } public record PageResult(List records, Long total, int current, int size, int pages, int sourceCurrent, int sourceSize, int sourceCode, Instant observedAt, AttemptMetadata egress) { diff --git a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java index e87136021..ce9d64d6b 100644 --- a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java +++ b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java @@ -304,12 +304,23 @@ public final class LingxingWmsClient { int status = requiredInt(record.get("status"), "STATUS"); if (exactDeliveryNo == null && status != 100) throw invalidResponse("DISCOVERY_STATUS_MISMATCH", 200); return new OrderSummary(deliveryNo, optionalText(record, "sourceNo"), requiredText(record, "customerCode", 64), - requiredText(record, "whCode", 64), status, optionalText(record, "createTime"), optionalText(record, "outboundTime")); + requiredText(record, "whCode", 64), status, optionalText(record, "createTime"), optionalText(record, "outboundTime"), listSkus(record.get("productList"))); }); return new PageResult(page.records(), page.total(), current, size, page.pages(), page.sourceCurrent(), page.sourceSize(), 200, page.observedAt(), metadata); } + private List listSkus(JsonNode products) { + if (products == null || products.isNull()) return null; + if (!products.isArray()) throw invalidResponse("PRODUCT_LIST_TYPE_INVALID", 200); + List items = new ArrayList<>(); + for (JsonNode item : products) { + if (!item.isObject()) throw invalidResponse("PRODUCT_LIST_ROW_INVALID", 200); + items.add(new ListSku(optionalText(item, "productSku"), optionalText(item, "productName"), quantity(item.get("qty")))); + } + return List.copyOf(items); + } + private QueryPage mappedPage(JsonNode data, int current, int size, String currentField, String sizeField, boolean outbound, AttemptMetadata metadata, Function projection) { JsonNode records = data.get("records"); diff --git a/integration-service/src/main/resources/openapi/lingxing-wms-v1.yaml b/integration-service/src/main/resources/openapi/lingxing-wms-v1.yaml index e52a65bcb..9ebf851be 100644 --- a/integration-service/src/main/resources/openapi/lingxing-wms-v1.yaml +++ b/integration-service/src/main/resources/openapi/lingxing-wms-v1.yaml @@ -349,6 +349,16 @@ components: status: {type: integer} createTime: {type: string, nullable: true} outboundTime: {type: string, nullable: true} + productList: + type: array + nullable: true + description: Per-SKU list quantities; missing items, SKU or qty require one detail supplement for this order. Zero is a known quantity. + items: + type: object + properties: + productSku: {type: string, nullable: true} + productName: {type: string, nullable: true} + qty: {type: number, nullable: true} Warehouse: type: object required: [whCode, warehouseType, status] diff --git a/integration-service/src/test/java/com/baiyue/integration/controller/LingxingInternalControllerTest.java b/integration-service/src/test/java/com/baiyue/integration/controller/LingxingInternalControllerTest.java index a911cc96d..a6de263c6 100644 --- a/integration-service/src/test/java/com/baiyue/integration/controller/LingxingInternalControllerTest.java +++ b/integration-service/src/test/java/com/baiyue/integration/controller/LingxingInternalControllerTest.java @@ -131,7 +131,7 @@ class LingxingInternalControllerTest { void authenticatedReadReturnsOnlyOutboundSummary() throws Exception { when(client.queryPage(any(), any())).thenReturn(new PageResult( List.of(new OrderSummary("TEST-OUT-001", "TEST-SRC-001", "TEST-CUSTOMER", "TEST-WH", 100, - "2026-10-01 08:00:00", "2026-10-02 08:00:00")), + "2026-10-01 08:00:00", "2026-10-02 08:00:00", null)), 1L, 1, 1, 1, 1, 1, 200, Instant.parse("2026-10-12T00:00:00Z"), null)); mvc.perform(post(PATH).header("X-Internal-Service-Token", token).header("X-Lingxing-Egress-Context", CONTEXT) .contentType(MediaType.APPLICATION_JSON).content(BODY)) diff --git a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java index d5b1deda4..2ba24e33a 100644 --- a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java +++ b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java @@ -81,6 +81,17 @@ class LingxingWmsClientTest { assertThat(result.observedAt()).isEqualTo(NOW); assertThat(result.pages()).isEqualTo(1); } + @Test void listProductsPreserveDecimalsZeroAndMissingQuantity() { + response.set(page("current","size",1,1,1,1,ORDER.strip().replace("}",""" + ,"productList":[{"productSku":"SKU","qty":2.123456},{"productSku":"ZERO","qty":0}, + {"productSku":"UNKNOWN"},{"qty":4}]}"""))); + var products=client().queryPage(query(1,1),context()).records().get(0).productList(); + assertThat(products).hasSize(4); + assertThat(products.get(0).qty()).isEqualByComparingTo("2.123456"); + assertThat(products.get(1).qty()).isEqualByComparingTo("0"); + assertThat(products.get(2).qty()).isNull();assertThat(products.get(3).productSku()).isNull(); + assertThat(calls.get()).isEqualTo(1); + } @Test void cancelledLocateOmitsTimeAndStatusAndRejectsFuzzyMatch() throws Exception { response.set(page("current", "size", 1, 1, 1, 1, ORDER.replace("100", "99").replace("2026-10-01 01:00:00", ""))); var observer = org.mockito.Mockito.mock(LingxingQuantityCapture.class); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/LingxingAnalyticsClient.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/LingxingAnalyticsClient.java index 0c44fb887..a492f4c2b 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/LingxingAnalyticsClient.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/LingxingAnalyticsClient.java @@ -21,5 +21,15 @@ public interface LingxingAnalyticsClient { Integer status, String timeZone) { } record Summary(String sourceAccount, String customerCode, String warehouseCode, String deliveryNo, String sourceNo, Integer sourceStatus, - String sourceCreatedTime, String sourceOutboundTime) { } + String sourceCreatedTime, String sourceOutboundTime, List items) { + public Order order() { + return new Order(sourceAccount, customerCode, warehouseCode, deliveryNo, sourceNo, sourceStatus, + sourceCreatedTime, sourceOutboundTime, items); + } + } + + static boolean needsDetail(Order order) { + return order.items() == null || order.items().isEmpty() || order.items().stream().anyMatch(item -> + item == null || item.skuCode() == null || item.skuCode().isBlank() || item.actualQuantity() == null); + } } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/RestLingxingAnalyticsClient.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/RestLingxingAnalyticsClient.java index 7c677fd5c..69435ef77 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/RestLingxingAnalyticsClient.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/RestLingxingAnalyticsClient.java @@ -110,9 +110,20 @@ public class RestLingxingAnalyticsClient implements LingxingAnalyticsClient { } private Summary summary(String account, JsonNode n) { + List items = null; + JsonNode products = n.get("productList"); + if (products != null && !products.isNull()) { + if (!products.isArray()) throw failure("LIST_ITEMS_INVALID"); + items = new ArrayList<>(); + for (JsonNode item : products) { + if (!item.isObject()) throw failure("LIST_ITEM_INVALID"); + items.add(new SelfopPayloads.Item(optional(item, "productSku"), optional(item, "productName"), + quantity(item, "qty"), properties.getQuantityUnit())); + } + } return new Summary(account, required(n, "customerCode", 64), required(n, "whCode", 64), required(n, "deliveryNo", 128), optional(n, "sourceNo"), integer(n, "status"), - optional(n, "createTime"), optional(n, "outboundTime")); + optional(n, "createTime"), optional(n, "outboundTime"), items); } private void gate(SelfopRequestContext context) { diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRoutingManifest.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRoutingManifest.java index 06228e752..10abe92fe 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRoutingManifest.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRoutingManifest.java @@ -7,6 +7,7 @@ import java.util.Map; /** Durable directory cursor and effective immutable stage references. */ @Data public class SelfopRoutingManifest { + private int listPayloadVersion; private int scan = 1; private int nextPage = 1; private Integer pages; diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopBudget.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopBudget.java index 46752ce75..fa1ce6748 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopBudget.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopBudget.java @@ -12,6 +12,7 @@ public final class SelfopBudget { if(limit<=0||seconds<=0)throw new IllegalArgumentException("SELFOP_BUDGET_INVALID"); this.limit=limit;this.deadline=Instant.now().plusSeconds(seconds); } - public boolean take() {if(requests>=limit||!Instant.now().isBefore(deadline))return false;requests++;return true;} + public boolean hasTime() {return Instant.now().isBefore(deadline);} + public boolean take() {if(requests>=limit||!hasTime())return false;requests++;return true;} public int used(){return requests;} } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java index 11dec60f0..f7013b831 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java @@ -44,6 +44,13 @@ public class SelfopDiscoveryService { private void collect(SelfopSyncTask task,SelfopSyncRun run,SelfopLeaseService.Lease lease,SelfopBudget budget) { Query query=read(task.getQueryScopeJson(),Query.class); var manifest=task.getRoutingManifestJson()==null?new SelfopRoutingManifest():read(task.getRoutingManifestJson(),SelfopRoutingManifest.class); + if(manifest.getListPayloadVersion()==0) { + if(!manifest.isDirectoryComplete()) { + manifest.setScan(1);manifest.setNextPage(1);manifest.setPages(null);manifest.setTotal(null); + manifest.getObjects().clear();manifest.getFirstScan().clear(); + } + manifest.setListPayloadVersion(1); + } if(query.deliveryNo()!=null) {if(!collectLocated(task,run,lease,budget,query,manifest))return;} else { while(!manifest.isDirectoryComplete()) { @@ -60,7 +67,7 @@ public class SelfopDiscoveryService { String key=hash(List.of(task.getSourceAccount(),summary.customerCode(),summary.deliveryNo()));String content=hash(summary); if(manifest.getScan()==1){ require(!manifest.getFirstScan().containsKey(key),"SELFOP_DIRECTORY_DUPLICATE");manifest.getFirstScan().put(key,content); - var payload=new SelfopPayloads.Order(task.getSourceAccount(),summary.customerCode(),summary.warehouseCode(),summary.deliveryNo(),summary.sourceNo(),summary.sourceStatus(),summary.sourceCreatedTime(),summary.sourceOutboundTime(),null); + var payload=new SelfopPayloads.Order(task.getSourceAccount(),summary.customerCode(),summary.warehouseCode(),summary.deliveryNo(),summary.sourceNo(),summary.sourceStatus(),summary.sourceCreatedTime(),summary.sourceOutboundTime(),summary.items()); var stage=stage(task,run,payload,manifest.getNextPage(),null,"PENDING"); var prior=stages.selectOne(new QueryWrapper().eq("stage_batch_id",task.getId()).eq("object_type","OUTBOUND_ORDER").eq("object_key_hash",key)); if(prior==null)stages.insert(stage);else if("PENDING".equals(prior.getValidationStatus())){ @@ -87,6 +94,7 @@ public class SelfopDiscoveryService { } route(task,run,manifest,lease,false); for(var entry:manifest.getObjects().entrySet()) { + if(!budget.hasTime()){taskService.yield(task.getId(),run.getId(),lease);return;} var ref=entry.getValue(); if(ref.getRefreshTaskId()!=null){ var refresh=tasks.selectById(ref.getRefreshTaskId());require(refresh!=null,"SELFOP_REFRESH_TASK_MISSING"); @@ -105,7 +113,6 @@ public class SelfopDiscoveryService { var effective=ref.getEffectiveStageId()==null?null:stages.selectById(ref.getEffectiveStageId()); boolean newer=latest!=null&&(effective==null?latest.getObservationSequence()>task.getObservationBaseline():latest.getObservationSequence()>effective.getObservationSequence()); if(effective!=null&&!newer) continue; - if(!budget.take()){taskService.yield(task.getId(),run.getId(),lease);return;} leases.renew(lease); SelfopSyncTask fetchTask=task;SelfopSyncRun fetchRun=run; if(newer){ @@ -118,25 +125,47 @@ public class SelfopDiscoveryService { fetchRun=taskService.begin(fetchTask.getId(),lease); fetchTask.setAttemptCount(fetchRun.getAttemptNo());fetchTask.setStatus("RUNNING"); } - long observation=leases.nextObservation(lease); - SelfopPayloads.Order detail; - try {detail=client.detail(SelfopRequestContext.forTask(fetchTask,fetchRun,config),ref.getWarehouseCode(),ref.getCustomerCode(),ref.getDeliveryNo());} - catch(AnalyticsClientException staleIdentity){ - if(!Set.of("DETAIL_IDENTITY_MISMATCH","INTEGRATION_DATA_MISSING","LINGXING_WMS_SOURCE_REJECTED","LINGXING_WMS_INVALID_RESPONSE").contains(staleIdentity.code())||staleIdentity.retryable())throw staleIdentity; - var relocate=new SelfopSyncTask();relocate.setSourceAccount(task.getSourceAccount());relocate.setTaskRole("ACCOUNT_DISCOVERY");relocate.setDiscoveryScopeId(task.getDiscoveryScopeId()); - relocate.setExecutionPurpose(task.getExecutionPurpose()); - relocate.setTaskType("RECHECK");relocate.setScopeHash(task.getScopeHash());relocate.setRequestKey("relocate:"+task.getId()+":"+entry.getKey().substring(0,24)+":"+observation); - var locateQuery=new Query(null,null,ref.getDeliveryNo(),ref.getCustomerCode(),task.getId());relocate.setQueryScopeJson(json(locateQuery));relocate.setRequestHash(hash(relocate.getQueryScopeJson())); - fetchTask=taskService.create(relocate);fetchRun=taskService.begin(fetchTask.getId(),lease);ref.setRefreshTaskId(fetchTask.getId()); - fetchTask.setAttemptCount(fetchRun.getAttemptNo());fetchTask.setStatus("RUNNING"); + if(newer){ tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);}); - var located=new SelfopRoutingManifest(); - if(!collectLocated(fetchTask,fetchRun,lease,budget,locateQuery,located)){tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);});taskService.yield(task.getId(),run.getId(),lease);return;} - ref.setWarehouseCode(located.getObjects().values().iterator().next().getWarehouseCode()); + collect(fetchTask,fetchRun,lease,budget); + var refreshedTask=tasks.selectById(fetchTask.getId()); + if(!"SUCCEEDED".equals(refreshedTask.getStatus())){taskService.yield(task.getId(),run.getId(),lease);return;} + var refreshed=read(refreshedTask.getRoutingManifestJson(),SelfopRoutingManifest.class).getObjects().get(entry.getKey()); + require(refreshed!=null&&refreshed.getEffectiveStageId()!=null,"SELFOP_REFRESH_REFERENCE_MISSING"); + ref.setEffectiveStageId(refreshed.getEffectiveStageId());ref.setWarehouseCode(refreshed.getWarehouseCode()); tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);}); - if(!budget.take()){taskService.yield(fetchTask.getId(),fetchRun.getId(),lease);taskService.yield(task.getId(),run.getId(),lease);return;} - observation=leases.nextObservation(lease);detail=client.detail(SelfopRequestContext.forTask(fetchTask,fetchRun,config),ref.getWarehouseCode(),ref.getCustomerCode(),ref.getDeliveryNo()); + continue; } + var directoryStage=ref.getDirectoryStageId()==null?null:stages.selectById(ref.getDirectoryStageId()); + SelfopPayloads.Order detail=directoryStage==null?null:read(directoryStage.getPayloadJson(),SelfopPayloads.Order.class); + long observation; + if(detail==null||LingxingAnalyticsClient.needsDetail(detail)) { + if(!budget.take()){taskService.yield(task.getId(),run.getId(),lease);return;} + observation=leases.nextObservation(lease); + try {detail=client.detail(SelfopRequestContext.forTask(fetchTask,fetchRun,config),ref.getWarehouseCode(),ref.getCustomerCode(),ref.getDeliveryNo());} + catch(AnalyticsClientException staleIdentity){ + if(!Set.of("DETAIL_IDENTITY_MISMATCH","INTEGRATION_DATA_MISSING","LINGXING_WMS_SOURCE_REJECTED","LINGXING_WMS_INVALID_RESPONSE").contains(staleIdentity.code())||staleIdentity.retryable())throw staleIdentity; + var relocate=new SelfopSyncTask();relocate.setSourceAccount(task.getSourceAccount());relocate.setTaskRole("ACCOUNT_DISCOVERY");relocate.setDiscoveryScopeId(task.getDiscoveryScopeId()); + relocate.setExecutionPurpose(task.getExecutionPurpose()); + relocate.setTaskType("RECHECK");relocate.setScopeHash(task.getScopeHash());relocate.setRequestKey("relocate:"+task.getId()+":"+entry.getKey().substring(0,24)+":"+observation); + var locateQuery=new Query(null,null,ref.getDeliveryNo(),ref.getCustomerCode(),task.getId());relocate.setQueryScopeJson(json(locateQuery));relocate.setRequestHash(hash(relocate.getQueryScopeJson())); + fetchTask=taskService.create(relocate);fetchRun=taskService.begin(fetchTask.getId(),lease);ref.setRefreshTaskId(fetchTask.getId()); + fetchTask.setAttemptCount(fetchRun.getAttemptNo());fetchTask.setStatus("RUNNING"); + tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);}); + var located=new SelfopRoutingManifest();located.setListPayloadVersion(1); + if(!collectLocated(fetchTask,fetchRun,lease,budget,locateQuery,located)){tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);});taskService.yield(task.getId(),run.getId(),lease);return;} + ref.setWarehouseCode(located.getObjects().values().iterator().next().getWarehouseCode()); + tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);}); + var locatedRef=located.getObjects().values().iterator().next(); + var locatedStage=stages.selectById(locatedRef.getDirectoryStageId()); + detail=read(locatedStage.getPayloadJson(),SelfopPayloads.Order.class); + if(LingxingAnalyticsClient.needsDetail(detail)) { + if(!budget.take()){taskService.yield(fetchTask.getId(),fetchRun.getId(),lease);taskService.yield(task.getId(),run.getId(),lease);return;} + detail=client.detail(SelfopRequestContext.forTask(fetchTask,fetchRun,config),ref.getWarehouseCode(),ref.getCustomerCode(),ref.getDeliveryNo()); + } + observation=leases.nextObservation(lease); + } + } else observation=leases.nextObservation(lease); require(Objects.equals(detail.sourceAccount(),task.getSourceAccount())&&Objects.equals(detail.customerCode(),ref.getCustomerCode()) &&Objects.equals(detail.deliveryNo(),ref.getDeliveryNo()),"SELFOP_DETAIL_IDENTITY_MISMATCH"); require(detail.items()!=null&&detail.sourceStatus()!=null,"SELFOP_DETAIL_STRUCTURE_INVALID"); @@ -163,7 +192,7 @@ public class SelfopDiscoveryService { }); route(task,run,manifest,lease,false); } - // A directory warehouse is provisional until every account detail (including resumed + // A directory warehouse is provisional until every account order (including resumed // revision checks) has resolved. An empty warehouse must not publish a false zero early. route(task,run,manifest,lease,true); tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);var currentTask=checkpoint.task();var currentRun=checkpoint.run();var now=utcNow(); @@ -191,7 +220,13 @@ public class SelfopDiscoveryService { require(Objects.equals(manifest.getTotal(),page.total())&&Objects.equals(manifest.getPages(),page.pages()),"SELFOP_PAGE_TOTAL_CHANGED"); for(var summary:page.records())if(query.deliveryNo().equals(summary.deliveryNo())&&query.customerCode().equals(summary.customerCode())){ String key=hash(List.of(task.getSourceAccount(),summary.customerCode(),summary.deliveryNo()));require(!manifest.getObjects().containsKey(key),"SELFOP_LOCATE_AMBIGUOUS"); - var e=new SelfopRoutingManifest.Entry();e.setCustomerCode(summary.customerCode());e.setDeliveryNo(summary.deliveryNo());e.setWarehouseCode(summary.warehouseCode());manifest.getObjects().put(key,e);} + var e=new SelfopRoutingManifest.Entry();e.setCustomerCode(summary.customerCode());e.setDeliveryNo(summary.deliveryNo());e.setWarehouseCode(summary.warehouseCode()); + tx.executeWithoutResult(s->{taskService.checkpoint(task,run,lease); + var pending=stage(task,run,summary.order(),manifest.getNextPage(),null,"PENDING"); + var prior=stages.selectOne(new QueryWrapper().eq("stage_batch_id",task.getId()).eq("object_type","OUTBOUND_ORDER").eq("object_key_hash",key)); + if(prior==null)stages.insert(pending);else pending=prior; + e.setDirectoryStageId(pending.getId());}); + manifest.getObjects().put(key,e);} manifest.setNextPage(manifest.getNextPage()+1);if(page.pages()==0||manifest.getNextPage()>page.pages()){ require(manifest.getObjects().size()==1,"SELFOP_LOCATE_UNRESOLVED");manifest.setDirectoryComplete(true);} tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);}); 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 4a2d84d10..1981528d2 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 @@ -88,7 +88,11 @@ public class SelfopValidationRunner { while(state.detailIndexclient.detail(context,summary.warehouseCode(),summary.customerCode(),summary.deliveryNo())); + SelfopPayloads.Order selected=summary.order(); + if(LingxingAnalyticsClient.needsDetail(selected)) { + selected=request(task,run,lease,()->client.detail(context,summary.warehouseCode(),summary.customerCode(),summary.deliveryNo())); + } + var detail=selected; validateOrder(task,summary,detail); tx.executeWithoutResult(s->{check(task,run,lease); boolean unknown=detail.items().stream().anyMatch(i->i.actualQuantity()==null); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/client/RestLingxingAnalyticsClientTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/client/RestLingxingAnalyticsClientTest.java index 7e9208102..cbd009edf 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/client/RestLingxingAnalyticsClientTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/client/RestLingxingAnalyticsClientTest.java @@ -74,6 +74,21 @@ class RestLingxingAnalyticsClientTest { assertThat(summary.warehouseCode()).isEqualTo("WH-B"); server.verify(); } + @Test void listItemsUseQtyAndKeepUnknownSeparateFromZero() { + enable();properties.setQuantityUnit("PCS"); + expect("/outbound-orders/locate", "{\"deliveryNo\":\"DO-1\",\"current\":1,\"size\":100}", """ + {"code":"200","data":{"current":1,"size":100,"pages":1,"total":1,"records":[ + {"deliveryNo":"DO-1","customerCode":"C","whCode":"WH-A","status":100, + "productList":[{"productSku":"SKU","qty":2.123456,"actualDeliveryQty":99}, + {"productSku":"ZERO","qty":0},{"productSku":"UNKNOWN"}]}]}} + """); + var summary=client.locate(context("synthetic-account"),"DO-1",1,100).records().get(0); + assertThat(summary.items().get(0).actualQuantity()).isEqualByComparingTo("2.123456"); + assertThat(summary.items().get(1).actualQuantity()).isEqualByComparingTo("0"); + assertThat(summary.items().get(2).actualQuantity()).isNull(); + assertThat(summary.items().get(0).quantityUnit()).isEqualTo("PCS"); + assertThat(LingxingAnalyticsClient.needsDetail(summary.order())).isTrue();server.verify(); + } @Test void detailProjectsActualLinesOnlyAndDoesNotInventUnitOrTimestamp() { enable(); expect("/outbound-orders/detail", "{\"warehouseCode\":\"WH-A\",\"customerCode\":\"C\",\"deliveryNo\":\"DO-1\"}", """ diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java index b971ca49e..7f7c5f187 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java @@ -102,7 +102,7 @@ class SelfopDiscoveryServiceTest { } private LingxingAnalyticsClient.Summary summary(String no, String time) { - return new LingxingAnalyticsClient.Summary("demo", "customer", "A", no, null, 100, null, time); + return new LingxingAnalyticsClient.Summary("demo", "customer", "A", no, null, 100, null, time, null); } private SelfopPayloads.Order detail(String no, String time) { return new SelfopPayloads.Order("demo", "customer", "A", no, null, 100, null, time, @@ -240,7 +240,7 @@ class SelfopDiscoveryServiceTest { }).when(taskService).yield(anyLong(), anyLong(), eq(lease)); when(client.detail(any(SelfopRequestContext.class), eq("A"), eq("customer"), eq("moved"))) .thenThrow(new AnalyticsClientException("LINGXING_WMS_SOURCE_REJECTED", false)); - var moved = new LingxingAnalyticsClient.Summary("demo", "customer", "B", "moved", null, 100, null, "2026-10-11 12:00:00"); + var moved = new LingxingAnalyticsClient.Summary("demo", "customer", "B", "moved", null, 100, null, "2026-10-11 12:00:00", null); when(client.locate(any(SelfopRequestContext.class), eq("moved"), eq(1), eq(2))).thenReturn(page(1, 2, 1, 1, moved)); when(client.detail(any(SelfopRequestContext.class), eq("B"), eq("customer"), eq("moved"))).thenReturn(new SelfopPayloads.Order("demo", "customer", "B", "moved", null, 100, null, "2026-10-11 12:00:00", List.of(new SelfopPayloads.Item("sku", null, BigDecimal.ONE, "PCS")))); @@ -296,6 +296,85 @@ class SelfopDiscoveryServiceTest { assertEquals("ROUTED",run.getStatus());assertEquals(500L,read(task.getRoutingManifestJson(),SelfopRoutingManifest.class).getObjects().get(key).getEffectiveStageId()); assertEquals(SelfopOutboundFingerprint.order(loaded),stored.get(500L).getContentHash()); } + private LingxingAnalyticsClient.Summary listed(String no, List items) { + var s=summary(no,"2026-10-11 12:00:00"); + return new LingxingAnalyticsClient.Summary(s.sourceAccount(),s.customerCode(),s.warehouseCode(),s.deliveryNo(), + s.sourceNo(),s.sourceStatus(),s.sourceCreatedTime(),s.sourceOutboundTime(),items); + } + @Test void completeListAndRealZeroNeedNoDetailsEvenWhenRequestBudgetIsExhausted() { + account.setRequestBudget(2); + when(client.discover(any(),eq(from),eq(until),eq(1),eq(2))).thenReturn(page(1,2,2,1, + listed("one",List.of(new SelfopPayloads.Item("sku",null,new BigDecimal("2.123456"),"PCS"))), + listed("zero",List.of(new SelfopPayloads.Item("sku",null,BigDecimal.ZERO,"PCS"))))); + service.collect(task,run,lease); + verify(client,never()).detail(any(),anyString(),anyString(),anyString()); + assertEquals("ROUTED",run.getStatus());assertEquals(2,stored.size()); + assertTrue(stored.values().stream().allMatch(s->"VALID".equals(s.getValidationStatus()))); + var quantities=stored.values().stream().map(s->read(s.getPayloadJson(),SelfopPayloads.Order.class).items().get(0).actualQuantity()).toList(); + assertTrue(quantities.stream().anyMatch(q->q.compareTo(new BigDecimal("2.123456"))==0)); + assertTrue(quantities.stream().anyMatch(q->q.signum()==0)); + } + @Test void completeListYieldsAtTimeLimitAndResumesWithoutRequests() { + when(client.discover(any(),eq(from),eq(until),eq(1),eq(2))).thenReturn(page(1,2,2,1, + listed("one",List.of(new SelfopPayloads.Item("sku",null,BigDecimal.ONE,"PCS"))), + listed("two",List.of(new SelfopPayloads.Item("sku",null,BigDecimal.TEN,"PCS"))))); + try(var budgets=mockConstruction(SelfopBudget.class,(budget,context)->{ + when(budget.take()).thenReturn(true); + when(budget.hasTime()).thenReturn(true,false); + })) { + service.collect(task,run,lease); + verify(budgets.constructed().get(0),times(2)).take(); + } + verify(taskService).yield(task.getId(),run.getId(),lease); + assertEquals(1,stored.values().stream().filter(s->"VALID".equals(s.getValidationStatus())).count()); + assertTrue(children.values().stream().allMatch(s->"WAITING_SOURCE".equals(s.getStatus()))); + service.collect(task,run,lease); + verify(client,times(2)).discover(any(),eq(from),eq(until),eq(1),eq(2)); + verify(client,never()).detail(any(),anyString(),anyString(),anyString()); + assertEquals(2,stored.values().stream().filter(s->"VALID".equals(s.getValidationStatus())).count()); + assertEquals("ROUTED",run.getStatus()); + } + @Test void onlyIncompleteOrderFallsBackAndRetryDoesNotRefetchCompletedOrders() { + when(client.discover(any(),eq(from),eq(until),eq(1),eq(2))).thenReturn(page(1,2,2,1, + listed("complete",List.of(new SelfopPayloads.Item("sku",null,BigDecimal.TEN,"PCS"))), + listed("missing",List.of(new SelfopPayloads.Item("sku",null,null,"PCS"))))); + when(client.detail(any(),eq("A"),eq("customer"),eq("missing"))) + .thenThrow(new AnalyticsClientException("INTEGRATION_PROVIDER_FAILURE",true)) + .thenReturn(detail("missing","2026-10-11 12:00:00")); + assertThrows(AnalyticsClientException.class,()->service.collect(task,run,lease)); + assertEquals(1,stored.values().stream().filter(s->"VALID".equals(s.getValidationStatus())).count()); + service.collect(task,run,lease); + verify(client,never()).detail(any(),eq("A"),eq("customer"),eq("complete")); + verify(client,times(2)).detail(any(),eq("A"),eq("customer"),eq("missing")); + verify(client,times(2)).discover(any(),eq(from),eq(until),eq(1),eq(2)); + assertEquals("ROUTED",run.getStatus()); + } + @Test void missingSkuOrEmptyItemsFallsBackPerOrder() { + when(client.discover(any(),eq(from),eq(until),eq(1),eq(2))).thenReturn(page(1,2,2,1, + listed("blank",List.of(new SelfopPayloads.Item(" ",null,BigDecimal.ONE,"PCS"))),listed("empty",List.of()))); + service.collect(task,run,lease); + verify(client).detail(any(),eq("A"),eq("customer"),eq("blank")); + verify(client).detail(any(),eq("A"),eq("customer"),eq("empty")); + assertEquals("ROUTED",run.getStatus()); + } + @Test void genuinelyEmptyDirectoryDoesNotFetchDetails() { + when(client.discover(any(),eq(from),eq(until),eq(1),eq(2))).thenReturn(page(1,2,0,0)); + service.collect(task,run,lease); + verify(client,never()).detail(any(),anyString(),anyString(),anyString()); + assertEquals("ROUTED",run.getStatus());assertTrue(stored.isEmpty()); + } + @Test void partialOldDirectoryRestartsScansWithListItems() { + var old=new SelfopRoutingManifest();old.setScan(2);old.setNextPage(1);old.setPages(1);old.setTotal(1L); + old.getFirstScan().put(hash(List.of("demo","customer","one")),"old-summary-without-items"); + task.setRoutingManifestJson(json(old)); + when(client.discover(any(),eq(from),eq(until),eq(1),eq(2))).thenReturn(page(1,2,1,1, + listed("one",List.of(new SelfopPayloads.Item("sku",null,BigDecimal.ONE,"PCS"))))); + service.collect(task,run,lease); + verify(client,times(2)).discover(any(),eq(from),eq(until),eq(1),eq(2)); + verify(client,never()).detail(any(),anyString(),anyString(),anyString()); + assertEquals("ROUTED",run.getStatus()); + assertEquals(1,read(task.getRoutingManifestJson(),SelfopRoutingManifest.class).getListPayloadVersion()); + } @Test void validationPurposeCannotReachAccountRoutingOrFormalCoverage() { task.setExecutionPurpose("VALIDATION");run.setExecutionPurpose("VALIDATION"); assertEquals("SELFOP_BUSINESS_PURPOSE_REQUIRED",assertThrows(IllegalStateException.class,()->service.collect(task,run,lease)).getMessage()); 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 71b66aeef..ab291976f 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 @@ -50,7 +50,7 @@ class SelfopValidationRunnerTest { } private LingxingAnalyticsClient.Page page(int current,long total,int pages,List records){return new LingxingAnalyticsClient.Page<>(records,current,10,total,pages);} 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");} + private LingxingAnalyticsClient.Summary summary(String number){return new LingxingAnalyticsClient.Summary("demo","customer","A",number,"source",100,null,"2026-10-12 08:02:00", null);} @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)))); @@ -186,7 +186,7 @@ class SelfopValidationRunnerTest { 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"); + var other=new LingxingAnalyticsClient.Summary("demo","customer","B","other","source",100,null,"2026-10-12 08:02:00", null); 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()));