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

feat(selfop): prefer list SKU quantities with targeted detail fallback

Browse Source
This commit is contained in:
ouyangxq
2026-10-14 08:15:16 +08:00
parent 4ae158c93e
commit c640fa2d0d
14 changed files with 220 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/LingxingAnalyticsClient.java
+11 -1
View File
@@ -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<SelfopPayloads.Item> 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);
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/RestLingxingAnalyticsClient.java
+12 -1
View File
@@ -110,9 +110,20 @@ public class RestLingxingAnalyticsClient implements LingxingAnalyticsClient {
}
private Summary summary(String account, JsonNode n) {
List<SelfopPayloads.Item> 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) {
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRoutingManifest.java
+1
View File
@@ -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;
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopBudget.java
+2 -1
View File
@@ -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;}
}
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java
+55 -20
View File
@@ -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<SelfopSyncStage>().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<SelfopSyncStage>().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);});
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopValidationRunner.java
+5 -1
View File
@@ -88,7 +88,11 @@ public class SelfopValidationRunner {
while(state.detailIndex<state.orders.size()) {
var summary=state.orders.get(state.detailIndex);leases.renew(lease);
require(selectedWarehouse.equals(summary.warehouseCode()),"SELFOP_SAMPLE_WAREHOUSE_MISMATCH");
var detail=request(task,run,lease,()->client.detail(context,summary.warehouseCode(),summary.customerCode(),summary.deliveryNo()));
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);