feat(selfop): sync versioned owner and product directories

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/LingxingAnalyticsClient.java

@@ -13,6 +13,8 @@ public interface LingxingAnalyticsClient {
     Page<Summary> locate(SelfopRequestContext context, String deliveryNo, int current, int size);
     Order detail(SelfopRequestContext context, String warehouseCode, String customerCode, String deliveryNo);
     Page<Stock> inventory(SelfopRequestContext context, String warehouseCode, int current, int size);
+    Page<SelfopPayloads.Owner> customers(SelfopRequestContext context, int current, int size);
+    Page<SelfopPayloads.Product> products(SelfopRequestContext context, int current, int size);
 
     record Page<T>(List<T> records, Integer current, Integer size, Long total, Integer pages) {
         public Page { records = List.copyOf(records); }

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/client/RestLingxingAnalyticsClient.java

@@ -109,6 +109,19 @@ public class RestLingxingAnalyticsClient implements LingxingAnalyticsClient {
         });
     }
 
+    @Override public Page<SelfopPayloads.Owner> customers(SelfopRequestContext context,int current,int size) {
+        gate(context);paging(current,size,1000);
+        return page(post(context,"/customers/query",Map.of("current",current,"size",size)),current,size,
+            n -> new SelfopPayloads.Owner(required(n,"customerCode",64),optional(n,"customerName"),optional(n,"status")));
+    }
+    @Override public Page<SelfopPayloads.Product> products(SelfopRequestContext context,int current,int size) {
+        gate(context);paging(current,size,100);
+        return page(post(context,"/products/query",Map.of("current",current,"size",size)),current,size,
+            n -> new SelfopPayloads.Product(required(n,"customerCode",64),required(n,"sku",128),optional(n,"productName"),
+                quantity(n,"length"),quantity(n,"width"),quantity(n,"height"),optional(n,"sizeUnit"),
+                quantity(n,"wmsLength"),quantity(n,"wmsWidth"),quantity(n,"wmsHeight"),optional(n,"wmsSizeUnit")));
+    }
+
     private Summary summary(String account, JsonNode n) {
         List<SelfopPayloads.Item> items = null;
         JsonNode products = n.get("productList");
@@ -145,6 +158,8 @@ public class RestLingxingAnalyticsClient implements LingxingAnalyticsClient {
             case "/outbound-orders/locate" -> Operation.OUTBOUND_LOCATE;
             case "/outbound-orders/detail" -> Operation.OUTBOUND_DETAIL;
             case "/product-inventory/query" -> Operation.PRODUCT_INVENTORY;
+            case "/customers/query" -> Operation.CUSTOMERS;
+            case "/products/query" -> Operation.PRODUCTS;
             default -> throw failure("EGRESS_OPERATION_INVALID");
         };
         int size = fields.containsKey("size") ? ((Number)fields.get("size")).intValue() : 0;

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/controller/PlatformAnalyticsController.java

@@ -64,6 +64,12 @@ public class PlatformAnalyticsController {
         @AuthenticationPrincipal AuthenticatedUser user,@ModelAttribute AnalyticsQuery query,@ModelAttribute AnalyticsInventoryOptions options){
         requirePlatform(user);return ApiResponse.success(service.owners(query,options));
     }
+    @GetMapping("/owner-directory")
+    @Operation(summary="读取固定查询时刻已发布的货主名称目录")
+    public ApiResponse<java.util.List<AnalyticsResponses.OwnerOption>> ownerDirectory(
+        @AuthenticationPrincipal AuthenticatedUser user,@ModelAttribute AnalyticsQuery query){
+        requirePlatform(user);return ApiResponse.success(service.ownerDirectory(query));
+    }
     @GetMapping("/owner-overview")
     @Operation(summary="本仓一个货主完整候选的30日经营概览")
     public ApiResponse<AnalyticsResponses.OwnerOverview> ownerOverview(

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopMasterDataMapper.java

@@ -0,0 +1,33 @@
+package com.baiyue.lastleg.selfoperated.analytics.mapper;
+
+import com.baiyue.lastleg.selfoperated.analytics.model.SelfopPayloads;
+import org.apache.ibatis.annotations.*;
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Mapper
+public interface SelfopMasterDataMapper {
+    @Insert("INSERT INTO ll_selfop_owner(id,run_id,source_account,customer_code,customer_name,source_status,observed_at_utc) VALUES(#{id},#{run},#{account},#{row.customerCode},#{row.customerName},#{row.status},#{now})")
+    void insertOwner(@Param("id") long id,@Param("run") long run,@Param("account") String account,@Param("row") SelfopPayloads.Owner row,@Param("now") LocalDateTime now);
+    @Insert("""
+        INSERT INTO ll_selfop_product(id,run_id,source_account,customer_code,sku_code,product_name,
+          declared_length,declared_width,declared_height,declared_size_unit,measured_length,measured_width,measured_height,measured_size_unit,observed_at_utc)
+        VALUES(#{id},#{run},#{account},#{row.customerCode},#{row.sku},#{row.productName},
+          #{row.length},#{row.width},#{row.height},#{row.sizeUnit},#{row.wmsLength},#{row.wmsWidth},#{row.wmsHeight},#{row.wmsSizeUnit},#{now})
+        """)
+    void insertProduct(@Param("id") long id,@Param("run") long run,@Param("account") String account,@Param("row") SelfopPayloads.Product row,@Param("now") LocalDateTime now);
+    @Select("SELECT COUNT(*) FROM ll_selfop_owner WHERE run_id=#{run} AND source_account=#{account}")
+    long ownerCount(@Param("run") long run,@Param("account") String account);
+    @Select("SELECT COUNT(*) FROM ll_selfop_product WHERE run_id=#{run} AND source_account=#{account}")
+    long productCount(@Param("run") long run,@Param("account") String account);
+    @Select("""
+        SELECT r.id FROM ll_selfop_sync_run r JOIN ll_selfop_sync_task t ON t.id=r.task_id
+        WHERE r.source_account=#{account} AND t.source_account=#{account} AND t.task_type='MASTER_DATA'
+          AND t.task_role='ACCOUNT_DISCOVERY' AND r.execution_purpose='BUSINESS' AND t.execution_purpose='BUSINESS'
+          AND r.status='PUBLISHED' AND r.coverage_status='COMPLETE' AND r.published_at_utc<=#{asOf}
+        ORDER BY r.published_at_utc DESC,r.id DESC LIMIT 1
+        """)
+    Long publishedAt(@Param("account") String account,@Param("asOf") LocalDateTime asOf);
+    @Select("SELECT customer_code,customer_name,source_status AS status FROM ll_selfop_owner WHERE source_account=#{account} AND run_id=#{run} ORDER BY customer_code")
+    List<SelfopPayloads.Owner> owners(@Param("account") String account,@Param("run") long run);
+}

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopCollectionPolicy.java

@@ -37,6 +37,15 @@ public final class SelfopCollectionPolicy {
         if(task==null||!"BUSINESS".equals(task.getExecutionPurpose())||task.getUpstreamTaskId()!=null
             ||task.getValidationRootTaskId()!=null||!CONTINUOUS_PLANNER.equals(task.getRequestedBy())||task.getScopeHash()==null||!task.getScopeHash().matches("[a-f0-9]{64}"))return false;
         String key=task.getRequestKey();if(key==null)return false;
+        if("MASTER_DATA".equals(task.getTaskType())) {
+            if(!"ACCOUNT_DISCOVERY".equals(task.getTaskRole())||task.getDiscoveryScopeId()==null||task.getScopeId()!=null
+                ||task.getWarehouseCode()!=null||task.getPlannedAtUtc()==null||task.getQueryFromUtc()!=null||task.getQueryUntilUtc()!=null)return false;
+            try{Map<?,?> q=read(task.getQueryScopeJson(),Map.class);
+                return key.equals("continuous:master:"+task.getPlannedAtUtc().plusHours(8).toLocalDate()+":"+task.getScopeHash().substring(0,12))
+                    &&q!=null&&q.size()==2&&"MASTER_DATA".equals(q.get("dataFamily"))&&task.getSourceAccount().equals(q.get("sourceAccount"))
+                    &&task.getScopeHash().equals(SelfopSupport.hash(new java.util.TreeMap<>(q)));
+            }catch(RuntimeException invalid){return false;}
+        }
         if("ACCOUNT_DISCOVERY".equals(task.getTaskRole())&&"BACKFILL".equals(task.getTaskType())&&task.getDiscoveryScopeId()!=null
             &&task.getQueryFromUtc()!=null&&task.getQueryUntilUtc()!=null&&task.getQueryUntilUtc().equals(task.getQueryFromUtc().plusDays(1))){
             var local=task.getQueryFromUtc().plusHours(8);

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopPayloads.java

@@ -5,6 +5,10 @@ import java.util.List;
 
 public final class SelfopPayloads {
     private SelfopPayloads() { }
+    public record Owner(String customerCode, String customerName, String status) { }
+    public record Product(String customerCode, String sku, String productName,
+        BigDecimal length, BigDecimal width, BigDecimal height, String sizeUnit,
+        BigDecimal wmsLength, BigDecimal wmsWidth, BigDecimal wmsHeight, String wmsSizeUnit) { }
     public record Order(String sourceAccount, String customerCode, String warehouseCode,
         String deliveryNo, String sourceNo, Integer sourceStatus, String sourceCreatedTime,
         String sourceOutboundTime, List<Item> items) { }

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopProgressProjection.java

@@ -11,7 +11,7 @@ import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*;
 public final class SelfopProgressProjection {
     private SelfopProgressProjection() { }
     private static final Set<String> PHASES = Set.of("WAITING_SOURCE", "DIRECTORY", "DETAIL", "INVENTORY",
-        "VALIDATION", "MERGE", "BUILD", "PUBLISH", "WAITING", "SAMPLE_DONE", "FAILED");
+        "VALIDATION", "MERGE", "BUILD", "PUBLISH", "WAITING", "SAMPLE_DONE", "FAILED", "CUSTOMERS", "PRODUCTS");
     /** Pure projection for callers already holding the scope/task/run locks and fence. */
     public static void applyCheckpoint(SelfopSyncTask task,SelfopSyncRun run,String phase,String unit,long completed,Long total,int scan) {
         applyCheckpoint(task,run,phase,unit,completed,total,scan,true);

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryService.java

@@ -21,17 +21,18 @@ public class AnalyticsQueryService {
     private final AnalyticsReadMapper mapper;
     private final ObjectMapper json;
     private final Clock clock;
+    private final SelfopOwnerDirectory ownerDirectory;
     @Value("${baiyue.self-operated-analytics.max-query-days:366}") private int maxQueryDays=366;
     @Value("${baiyue.self-operated-analytics.sales-retention-days:7}") private int retentionDays=7;
-    @Autowired public AnalyticsQueryService(AnalyticsReadMapper mapper,ObjectMapper json){this(mapper,json,Clock.systemUTC());}
-    AnalyticsQueryService(AnalyticsReadMapper mapper,ObjectMapper json,Clock clock){this.mapper=mapper;this.json=json;this.clock=clock;}
+    @Autowired public AnalyticsQueryService(AnalyticsReadMapper mapper,ObjectMapper json,SelfopOwnerDirectory ownerDirectory){this(mapper,json,Clock.systemUTC(),ownerDirectory);}
+    AnalyticsQueryService(AnalyticsReadMapper mapper,ObjectMapper json,Clock clock,SelfopOwnerDirectory ownerDirectory){this.mapper=mapper;this.json=json;this.clock=clock;this.ownerDirectory=ownerDirectory;}
     record Snapshot(AnalyticsQuery q,Run sales,Run inventory,boolean sourceKnown,boolean trusted,
         LocalDate fullFrom,LocalDate fullUntil,SalesRangePolicy.Range range,AnalyticsResponses.Context context){}
 
     public AnalyticsResponses.Page<AnalyticsResponses.SalesSummary> salesSummary(AnalyticsQuery q,SalesSort salesSort){
         Snapshot s=context(q,true,false,false,null); AnalyticsSqlRequest p=request(s,"SUMMARY");
         long count=mapper.count(p);
-        var records=mapper.salesPage(p,salesSort).stream().map(r->summary(s,r)).toList();
+        var names=ownerNames(s);var records=mapper.salesPage(p,salesSort).stream().map(r->summary(s,r,names.get(r.customerCode()))).toList();
         return page(s,records,count,Map.of());
     }
     public AnalyticsResponses.Trend dailySales(AnalyticsQuery q){
@@ -64,7 +65,7 @@ public class AnalyticsQueryService {
         var totals=SalesBucketPolicy.aggregate(evaluated);
         var summary=new AnalyticsResponses.SalesSummary(q.sourceAccount(),q.warehouseCode(),q.customerCode(),q.sku(),name(sp,ip),unit,
             totals.outboundQuantity(),totals.observedQuantity(),totals.dataStatus(),totals.reasonCodes(),
-            totals.completeDays(),totals.partialDays(),totals.unknownDays());
+            totals.completeDays(),totals.partialDays(),totals.unknownDays(),ownerNames(s).get(q.customerCode()));
         return new AnalyticsResponses.Trend(s.context,summary,List.copyOf(points));
     }
     public AnalyticsResponses.Page<AnalyticsResponses.Metric> hotSkus(AnalyticsQuery q){return metrics(q,"HOT",true);}
@@ -73,11 +74,13 @@ public class AnalyticsQueryService {
         o.validate("INVENTORY",q);Snapshot s=context(q,false,false,false,o);var p=request(s,"INVENTORY");
         CoverageDistributionPolicy.requireDrilldown(q,o,s.context(),inventoryFreshUntil(s),clock.instant());
         long total=mapper.inventoryCount(p,o);
-        return page(s,q.offset()>=total?List.of():mapper.inventoryPage(p,o).stream().map(r->inventoryMetric(s,r)).toList(),total,Map.of());
+        var names=ownerNames(s);
+        return page(s,q.offset()>=total?List.of():mapper.inventoryPage(p,o).stream().map(r->inventoryMetric(s,r,names.get(r.customerCode()))).toList(),total,Map.of());
     }
     public AnalyticsResponses.Page<AnalyticsResponses.OwnerOption> owners(AnalyticsQuery q,AnalyticsInventoryOptions o){
         o.validate("OWNERS",q);Snapshot s=context(q,false,false,false,null);var p=request(s,"OWNERS");long total=mapper.ownerCount(p,o);
-        return page(s,q.offset()>=total?List.of():mapper.owners(p,o).stream().map(code->new AnalyticsResponses.OwnerOption(code,null)).toList(),total,Map.of());
+        var names=ownerNames(s);
+        return page(s,q.offset()>=total?List.of():mapper.owners(p,o).stream().map(code->new AnalyticsResponses.OwnerOption(code,names.get(code))).toList(),total,Map.of());
     }
     public AnalyticsResponses.OwnerOverview ownerOverview(AnalyticsQuery q,AnalyticsInventoryOptions o){
         o.validate("OVERVIEW",q);Snapshot s=context(q,false,false,false,null);var raw=mapper.ownerTotals(request(s,"OVERVIEW"),o);
@@ -89,11 +92,11 @@ public class AnalyticsQueryService {
         var average=new AnalyticsResponses.QuantityTotal(divide30(outbound.overallTotal()),divide30(outbound.knownSubtotal()),outbound.unknownSkuCount(),outbound.dataStatus(),outbound.reasonCodes());
         var physicalSkus=new AnalyticsResponses.QuantityTotal(physical.unknownSkuCount()==0?decimal(totals,"physical_sku_known"):null,
             decimal(totals,"physical_sku_known"),physical.unknownSkuCount(),physical.dataStatus(),physical.reasonCodes());
-        return new AnalyticsResponses.OwnerOverview(s.context,q.customerCode(),null,count,physicalSkus,physical,quantity(totals,"available_quantity"),
+        return new AnalyticsResponses.OwnerOverview(s.context,q.customerCode(),ownerNames(s).get(q.customerCode()),count,physicalSkus,physical,quantity(totals,"available_quantity"),
             quantity(totals,"frozen_quantity"),quantity(totals,"defective_physical_quantity"),quantity(totals,"transit_quantity"),outbound,average,
             risk(totals,"no_outbound",physical,false),risk(totals,"above_threshold",physical,o.coverageDaysThreshold()==null));
     }
-    private AnalyticsResponses.InventoryMetric inventoryMetric(Snapshot s,Inventory r){
+    private AnalyticsResponses.InventoryMetric inventoryMetric(Snapshot s,Inventory r,String customerName){
         boolean observed=Boolean.TRUE.equals(r.observationEligible());String quality=!"VALID".equals(s.context.inventoryStatus())?s.context.inventoryStatus():r.availableQuantity()==null?"UNKNOWN":"VALID";
         String coverage=!Boolean.TRUE.equals(r.unitValid())?"UNIT_UNKNOWN_OR_CONFLICT":r.ownerOutbound()!=null?"COMPLETE":"COVERAGE_INSUFFICIENT";
         var calculated=InventoryWindowPolicy.calculate(quality,r.availableQuantity(),r.outboundQuantity30d(),s.trusted,
@@ -115,7 +118,7 @@ public class AnalyticsQueryService {
         return new AnalyticsResponses.InventoryMetric(s.q.sourceAccount(),r.customerCode(),s.q.warehouseCode(),r.skuCode(),r.productName(),Boolean.TRUE.equals(r.unitValid())?"PCS":null,
             r.orderCount(),r.physicalQuantity(),r.availableQuantity(),r.frozenQuantity(),r.transitQuantity(),utc(r.observedAtUtc()),
             r.outboundQuantity30d()==null?null:r.outboundQuantity30d().divide(BigDecimal.valueOf(s.range.dayCount()),12,java.math.RoundingMode.HALF_UP),calculated.days(),quality,coverage,observed,utc(r.evidenceAtUtc()),calculated.status(),List.copyOf(new LinkedHashSet<>(reasons)),
-            null,r.defectivePhysicalQuantity(),r.outboundQuantity30d(),r.observedOutbound(),r.lastOutbound(),historyStart(s),List.copyOf(attention),Map.copyOf(statuses));
+            customerName,r.defectivePhysicalQuantity(),r.outboundQuantity30d(),r.observedOutbound(),r.lastOutbound(),historyStart(s),List.copyOf(attention),Map.copyOf(statuses));
     }
     private static boolean positive(BigDecimal n){return n!=null&&n.signum()>0;}
     private static LocalDate historyStart(Snapshot s){return SalesQueryCoverage.intervals(s.sales).stream()
@@ -141,21 +144,26 @@ public class AnalyticsQueryService {
         Snapshot s=context(q,false,false,hot,null); AnalyticsSqlRequest p=request(s,kind);
         List<AnalyticsReadMapper.RankingStat> statistics=mapper.rankingStats(p);
         long total=statistics.isEmpty()?0:statistics.get(0).total();
-        List<AnalyticsResponses.Metric> rows=q.offset()>=total?List.of():mapper.page(p).stream().map(r->metric(s,r,kind)).toList();
+        var names=ownerNames(s);List<AnalyticsResponses.Metric> rows=q.offset()>=total?List.of():mapper.page(p).stream().map(r->metric(s,r,kind,names.get(r.customerCode()))).toList();
         Map<String,Long> excluded=new TreeMap<>();
         for(var e:statistics)if(e.reason()!=null)excluded.put(e.reason(),e.excludedCount());
         return page(s,rows,total,excluded);
     }
-    private AnalyticsResponses.SalesSummary summary(Snapshot s,Aggregate r){
+    private Map<String,String> ownerNames(Snapshot s){return ownerDirectory.names(s.q.sourceAccount(),s.context.queryAsOf());}
+    public List<AnalyticsResponses.OwnerOption> ownerDirectory(AnalyticsQuery q){
+        Snapshot s=context(q,false,false,false,null);
+        return ownerNames(s).entrySet().stream().map(e->new AnalyticsResponses.OwnerOption(e.getKey(),e.getValue())).toList();
+    }
+    private AnalyticsResponses.SalesSummary summary(Snapshot s,Aggregate r,String customerName){
         int complete=r.completeDayCount(),partial=r.partialDayCount(),unknown=r.unknownDayCount();
         String status=complete==s.range.dayCount()?"COMPLETE":r.observedQuantity()!=null?"PARTIAL":"UNKNOWN";
         List<String> reasons=new ArrayList<>(s.context.reasonCodes());
         if(!Boolean.TRUE.equals(r.unitValid()))reasons.add("UNIT_UNKNOWN_OR_CONFLICT");
         if(partial>0)reasons.add("SALES_DAY_INCOMPLETE"); if(unknown>0)reasons.add("SALES_NOT_COVERED_OR_SKU_HISTORY_UNKNOWN");
         return new AnalyticsResponses.SalesSummary(s.q.sourceAccount(),s.q.warehouseCode(),r.customerCode(),r.skuCode(),r.productName(),
-            r.quantityUnit(),"COMPLETE".equals(status)?r.observedQuantity():null,r.observedQuantity(),status,List.copyOf(reasons),complete,partial,unknown);
+            r.quantityUnit(),"COMPLETE".equals(status)?r.observedQuantity():null,r.observedQuantity(),status,List.copyOf(reasons),complete,partial,unknown,customerName);
     }
-    private AnalyticsResponses.Metric metric(Snapshot s,Aggregate r,String kind){
+    private AnalyticsResponses.Metric metric(Snapshot s,Aggregate r,String kind,String customerName){
         Instant evidence=utc(r.evidenceAtUtc());
         boolean observed=evidence!=null&&!evidence.isAfter(s.range.startDate().atStartOfDay(BUSINESS_ZONE).toInstant());
         boolean stockValid="VALID".equals(r.inventoryQuality());
@@ -178,7 +186,7 @@ public class AnalyticsQueryService {
         return new AnalyticsResponses.Metric(s.q.sourceAccount(),r.customerCode(),s.q.warehouseCode(),r.skuCode(),r.productName(),r.quantityUnit(),
             r.windowQuantity(),r.windowSkuOrderCount(),today,stockValid?r.physicalQuantity():null,stockValid?r.availableQuantity():null,
             stockValid?r.frozenQuantity():null,stockValid?r.transitQuantity():null,utc(r.inventoryObservedAtUtc()),result.average(),result.days(),
-            r.inventoryQuality(),coverage,observed,evidence,result.status(),List.copyOf(new LinkedHashSet<>(reasons)),null,null,r.sales30Quantity(),null,historyStart(s),List.of(),Map.of());
+            r.inventoryQuality(),coverage,observed,evidence,result.status(),List.copyOf(new LinkedHashSet<>(reasons)),customerName,null,r.sales30Quantity(),null,historyStart(s),List.of(),Map.of());
     }
     Snapshot context(AnalyticsQuery q,boolean chart,boolean exact,boolean hot,AnalyticsInventoryOptions inventoryOptions){
         boolean dashboardWindow=inventoryOptions!=null&&"DASHBOARD".equals(inventoryOptions.drilldownFrom());

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsResponses.java

@@ -17,7 +17,7 @@ public final class AnalyticsResponses {
     public record Page<T>(Context context,List<T> records,int current,int size,long total,long pages,Map<String,Long> excludedReasons){}
     public record SalesSummary(String sourceAccount,String warehouseCode,String customerCode,String sku,String productName,
         String quantityUnit,BigDecimal outboundQuantity,BigDecimal observedQuantity,String dataStatus,List<String> reasonCodes,
-        int completeDayCount,int partialDayCount,int unknownDayCount){}
+        int completeDayCount,int partialDayCount,int unknownDayCount,String customerName){}
     public record Point(LocalDate businessDate,BigDecimal outboundQuantity,BigDecimal observedQuantity,String dataStatus,
         List<String> reasonCodes,String quantityUnit,Instant coveredFrom,Instant coveredUntil){}
     public record Trend(Context context,SalesSummary summary,List<Point> points){}

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java

@@ -39,7 +39,7 @@ public interface CollectionReadMapper {
             + " AND JSON_TYPE(JSON_EXTRACT(t.query_scope_json,'$.warehouses[0]'))='STRING')";
     String TASK_WAREHOUSE = "CASE WHEN " + VALIDATION_ROOT
             + " THEN JSON_UNQUOTE(JSON_EXTRACT(t.query_scope_json,'$.warehouses[0]')) ELSE t.warehouse_code END";
-    String TASK_FAMILY = "CASE WHEN t.task_role='ACCOUNT_DISCOVERY' THEN 'OUTBOUND' ELSE s.data_family END";
+    String TASK_FAMILY = "CASE WHEN t.task_type='MASTER_DATA' THEN 'MASTER_DATA' WHEN t.task_role='ACCOUNT_DISCOVERY' THEN 'OUTBOUND' ELSE s.data_family END";
     String TASK_MEMBERSHIP = "((t.task_role='WAREHOUSE_SYNC' AND s.id IS NOT NULL) OR " + VALIDATION_ROOT + ")";
     String BUSINESS_LOCAL_WORK_READY = "CASE WHEN t.task_role='WAREHOUSE_SYNC' AND s.data_family='OUTBOUND'"
             + " AND (t.task_type='REBUILD' OR t.apply_state='APPLIED' OR ("

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardQueryService.java

@@ -14,10 +14,12 @@ import java.util.*;
 public class DashboardQueryService {
     private final AnalyticsQueryService contextService;
     private final DashboardReadMapper mapper;
-    public DashboardQueryService(AnalyticsQueryService contextService,DashboardReadMapper mapper){this.contextService=contextService;this.mapper=mapper;}
+    private final SelfopOwnerDirectory ownerDirectory;
+    public DashboardQueryService(AnalyticsQueryService contextService,DashboardReadMapper mapper,SelfopOwnerDirectory ownerDirectory){this.contextService=contextService;this.mapper=mapper;this.ownerDirectory=ownerDirectory;}
     public DashboardResponses.Dashboard dashboard(AnalyticsQuery input){
         AnalyticsQuery q=DashboardRequest.normalize(input);
         var snapshot=contextService.context(q,true,false,false,null);var c=snapshot.context();
+        var names=ownerDirectory.names(q.sourceAccount(),c.queryAsOf());
         if(snapshot.sales()!=null&&!Boolean.TRUE.equals(snapshot.sales().warehouseDailyReady()))
             throw new com.baiyue.common.core.exception.BizException("该销量版本尚未生成仓库每日汇总，请重新发布统计后再查询");
         var base=contextService.request(snapshot,"DASHBOARD");LocalDate windowFrom=c.rangeStartDate(),windowUntil=c.rangeEndDateExclusive();
@@ -38,7 +40,7 @@ public class DashboardQueryService {
             hotStats=hotReport.get(0);hotRanked=hotReport.stream().filter(row->row.get("rank_code")!=null).map(DashboardQueryService::ranked).toList();
         }
         var salesSummary=new AnalyticsResponses.SalesSummary(q.sourceAccount(),q.warehouseCode(),null,null,null,"PCS",trendSummary.outboundQuantity(),trendSummary.observedQuantity(),
-            trendSummary.dataStatus(),trendSummary.reasonCodes(),trendSummary.completeDays(),trendSummary.partialDays(),trendSummary.unknownDays());
+            trendSummary.dataStatus(),trendSummary.reasonCodes(),trendSummary.completeDays(),trendSummary.partialDays(),trendSummary.unknownDays(),null);
         var points=trendPoints.stream().map(t->new AnalyticsResponses.Point(t.date(),t.outboundQuantity(),t.observedQuantity(),t.dataStatus(),t.reasonCodes(),"PCS",t.provedFrom(),t.provedUntil())).toList();
         var interval=new DashboardResponses.Metric(trendSummary.outboundQuantity(),trendSummary.observedQuantity(),"PCS",trendSummary.dataStatus(),null,trendSummary.reasonCodes(),trendSummary.outboundQuantity(),trendSummary.observedQuantity());
         // Partial sales never become a formal daily average for the selected window.
@@ -51,14 +53,14 @@ public class DashboardQueryService {
         Instant freshUntil=contextService.inventoryFreshUntil(snapshot),drilldownUntil=CoverageDistributionPolicy.expiresAt(c,freshUntil);
         return new DashboardResponses.Dashboard(new DashboardResponses.Context(c,freshUntil,drilldownUntil),kpis,
             new DashboardResponses.SalesTrend(salesSummary,points),distribution(stats,inventoryAvailable,c),
-            hotRanking(ranking("HOT",hotStats,hotRanked,base.salesRun()>0,c.rangeStartDate(),c.rangeEndDateExclusive(),c),trendSummary.dataStatus()),
-            ranking("LONG",stats,ranked,inventoryAvailable,windowFrom,windowUntil,c),
-            ranking("NO_OUTBOUND",stats,ranked,inventoryAvailable,windowFrom,windowUntil,c));
+            hotRanking(ranking("HOT",hotStats,hotRanked,base.salesRun()>0,c.rangeStartDate(),c.rangeEndDateExclusive(),c,names),trendSummary.dataStatus()),
+            ranking("LONG",stats,ranked,inventoryAvailable,windowFrom,windowUntil,c,names),
+            ranking("NO_OUTBOUND",stats,ranked,inventoryAvailable,windowFrom,windowUntil,c,names));
     }
     private static DashboardResponses.Ranking hotRanking(DashboardResponses.Ranking ranking,String windowStatus){
         if("COMPLETE".equals(windowStatus))return ranking;
         var items=ranking.items().stream().map(i->new DashboardResponses.Item(i.customerCode(),i.sku(),i.productName(),i.value(),i.unit(),i.windowOutboundQuantity(),i.skuOutboundOrderCount(),
-            i.physicalQuantity(),i.availableQuantity(),i.frozenQuantity(),i.defectivePhysicalQuantity(),null,null)).toList();
+            i.physicalQuantity(),i.availableQuantity(),i.frozenQuantity(),i.defectivePhysicalQuantity(),null,null,i.customerName())).toList();
         var reasons=new LinkedHashSet<>(ranking.reasonCodes());reasons.add("SALES_WINDOW_INCOMPLETE");
         return new DashboardResponses.Ranking(ranking.windowStartDate(),ranking.windowEndDateExclusive(),ranking.calculationWindowDays(),ranking.eligibleCount(),ranking.unassessableCount(),items,
             items.isEmpty()&&"UNKNOWN".equals(windowStatus)?"UNKNOWN":"PARTIAL",List.copyOf(reasons));
@@ -86,7 +88,7 @@ public class DashboardQueryService {
         return new DashboardResponses.Distribution(CoverageDistributionPolicy.RULE_VERSION,available?count(s,"candidate_skus"):null,
             available?count(s,"excluded_zero_skus"):null,List.copyOf(buckets),status,codes,available?Map.copyOf(reasons):Map.of());
     }
-    private static DashboardResponses.Ranking ranking(String kind,Map<String,Object> s,List<DashboardReadMapper.Ranked> rows,boolean available,LocalDate from,LocalDate until,AnalyticsResponses.Context c){
+    private static DashboardResponses.Ranking ranking(String kind,Map<String,Object> s,List<DashboardReadMapper.Ranked> rows,boolean available,LocalDate from,LocalDate until,AnalyticsResponses.Context c,Map<String,String> names){
         int windowDays=(int)java.time.temporal.ChronoUnit.DAYS.between(from,until);
         String prefix=switch(kind){case "HOT"->"hot";case "LONG"->"long";default->"no_outbound";};
         List<DashboardReadMapper.Ranked> selected=rows.stream().filter(r->kind.equals(r.rankCode())).sorted(Comparator
@@ -95,7 +97,7 @@ public class DashboardQueryService {
             .thenComparing(DashboardReadMapper.Ranked::skuCode).thenComparing(DashboardReadMapper.Ranked::customerCode)).toList();
         var items=available?selected.stream().map(r->new DashboardResponses.Item(r.customerCode(),r.skuCode(),r.productName(),r.mainValue(),"LONG".equals(kind)?"DAY":"PCS",
             "HOT".equals(kind)?r.ownerOutbound():r.outboundQuantity30d(),r.orderCount(),r.physicalQuantity(),r.availableQuantity(),r.frozenQuantity(),r.defectivePhysicalQuantity(),
-            divide("HOT".equals(kind)?r.ownerOutbound():r.outboundQuantity30d(),windowDays),"HOT".equals(kind)?null:r.coverageDays())).toList():List.<DashboardResponses.Item>of();
+            divide("HOT".equals(kind)?r.ownerOutbound():r.outboundQuantity30d(),windowDays),"HOT".equals(kind)?null:r.coverageDays(),names.get(r.customerCode()))).toList():List.<DashboardResponses.Item>of();
         long unknown=count(s,prefix+"_unknown");String status=!available?"UNKNOWN":unknown>0?"PARTIAL":"COMPLETE";
         List<String> reasons=!available?List.of("HOT".equals(kind)?"SALES_UNAVAILABLE":"INVENTORY_UNAVAILABLE"):unknown>0?List.of("UNASSESSABLE_SKUS"):List.of();
         if(available&&count(s,prefix+"_eligible")==0&&unknown==0)reasons=List.of("NO_ELIGIBLE_SKUS");

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardResponses.java

@@ -20,7 +20,7 @@ public final class DashboardResponses {
         List<Bucket> buckets,String dataStatus,List<String> reasonCodes,Map<String,Long> unknownReasons){}
     public record Item(String customerCode,String sku,String productName,BigDecimal value,String unit,BigDecimal windowOutboundQuantity,
         Long skuOutboundOrderCount,BigDecimal physicalQuantity,BigDecimal availableQuantity,BigDecimal frozenQuantity,
-        BigDecimal defectivePhysicalQuantity,BigDecimal averageDailyOutboundQuantity,BigDecimal inventoryCoverageDays){}
+        BigDecimal defectivePhysicalQuantity,BigDecimal averageDailyOutboundQuantity,BigDecimal inventoryCoverageDays,String customerName){}
     public record Ranking(LocalDate windowStartDate,LocalDate windowEndDateExclusive,int calculationWindowDays,Long eligibleCount,
         Long unassessableCount,List<Item> items,String dataStatus,List<String> reasonCodes){}
     public record Dashboard(Context context,Kpis kpis,SalesTrend salesTrend,Distribution coverageDistribution,

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContext.java

@@ -18,11 +18,13 @@ public class InventoryTurnoverContext {
     private final InventoryTurnoverReadMapper mapper;
     private final ObjectMapper json;
     private final Clock clock;
-    @Autowired public InventoryTurnoverContext(AnalyticsReadMapper analytics,InventoryTurnoverReadMapper mapper,ObjectMapper json) {
-        this(analytics,mapper,json,Clock.systemUTC());
+    private final SelfopOwnerDirectory ownerDirectory;
+    @Autowired public InventoryTurnoverContext(AnalyticsReadMapper analytics,InventoryTurnoverReadMapper mapper,ObjectMapper json,SelfopOwnerDirectory ownerDirectory) {
+        this(analytics,mapper,json,Clock.systemUTC(),ownerDirectory);
     }
-    InventoryTurnoverContext(AnalyticsReadMapper analytics,InventoryTurnoverReadMapper mapper,ObjectMapper json,Clock clock) {
+    InventoryTurnoverContext(AnalyticsReadMapper analytics,InventoryTurnoverReadMapper mapper,ObjectMapper json,Clock clock,SelfopOwnerDirectory ownerDirectory) {
         this.analytics=analytics;this.mapper=mapper;this.json=json;this.clock=clock;
+        this.ownerDirectory=ownerDirectory;
     }
     public record Evidence(String account,String warehouse,LocalDate end,int days,Instant asOf,
         Long salesRun,Long inventoryRun,String salesHash,String inventoryHash,boolean rebuild,
@@ -41,7 +43,7 @@ public class InventoryTurnoverContext {
         var aq=new AnalyticsQuery(q.sourceAccount(),q.warehouseCode(),null,null,"CUSTOM",end.minusDays(Math.max(30,q.days())-1),end,"DAY",null,1,20,
             sales==null?0L:sales.id(),stock==null?0L:stock.id(),sales==null?null:sales.scopeHash(),stock==null?null:stock.scopeHash(),asOf);
         // Staleness is evaluated at queryAsOf, so pagination does not age a fixed snapshot midway.
-        var snapshot=new AnalyticsQueryService(analytics,json,Clock.fixed(asOf,ZoneOffset.UTC)).context(aq,true,false,false,null);
+        var snapshot=new AnalyticsQueryService(analytics,json,Clock.fixed(asOf,ZoneOffset.UTC),ownerDirectory).context(aq,true,false,false,null);
         String inventoryHash=scopes.stream().filter(s->"INVENTORY".equals(s.dataFamily())).map(AnalyticsReadRows.Scope::scopeHash).findFirst().orElse(null);
         var mapping=new TreeMap<LocalDate,Long>();
         for(var sample:mapper.samples(q.sourceAccount(),q.warehouseCode(),inventoryHash,utc(asOf),end.minusDays(Math.max(30,q.days())-1),end.plusDays(1))) {

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicy.java

@@ -65,7 +65,7 @@ public final class InventoryTurnoverPolicy {
         if(decimal(r,"defective")!=null&&decimal(r,"defective").signum()>0)attention.add("DEFECTIVE_STOCK");
         if(count(r,"unassessable")>0)attention.add("UNASSESSABLE");
         String last=text(r,"last_outbound");
-        return new Row(text(r,"customer_code"),null,owner?null:text(r,"sku_code"),owner?null:text(r,"product_name"),physical,available,
+        return new Row(text(r,"customer_code"),text(r,"customer_name"),owner?null:text(r,"sku_code"),owner?null:text(r,"product_name"),physical,available,
             total(r,"frozen",owner),total(r,"defective",owner),total(r,"transit",owner),owner?decimal(r,"physical_skus"):physical.value()==null?null:physical.value().signum()>0?BigDecimal.ONE:BigDecimal.ZERO,
             windows,base.averageDailyOutboundQuantity(),coverage,coverageStatus,average,turnover,sampleStatus,sampleCount,q.days(),List.copyOf(missing),
             "VALID".equals(sampleStatus)?List.of():List.of(sampleStatus),last==null?null:LocalDate.parse(last.substring(0,10)),List.copyOf(attention),

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverQueryService.java

@@ -16,9 +16,11 @@ public class InventoryTurnoverQueryService {
     private final InventoryTurnoverContext contexts;
     private final InventoryTurnoverReadMapper mapper;
     private final ObjectMapper json;
-    public InventoryTurnoverQueryService(InventoryTurnoverContext contexts,InventoryTurnoverReadMapper mapper,ObjectMapper json){this.contexts=contexts;this.mapper=mapper;this.json=json;}
+    private final SelfopOwnerDirectory ownerDirectory;
+    public InventoryTurnoverQueryService(InventoryTurnoverContext contexts,InventoryTurnoverReadMapper mapper,ObjectMapper json,SelfopOwnerDirectory ownerDirectory){this.contexts=contexts;this.mapper=mapper;this.json=json;this.ownerDirectory=ownerDirectory;}
     public Response query(InventoryTurnoverRequest request) {
         var frozen=contexts.resolve(request);var input=InventoryTurnoverSql.input(request,frozen);var context=frozen.response();
+        var names=ownerDirectory.names(request.sourceAccount(),context.queryAsOf());
         long total=0;boolean hasTotal=false;Overview summary=null;Charts charts=null;
         List<Row> rows=new ArrayList<>();List<OwnerPhysical> owners=new ArrayList<>();int unknown=0;
         for(var part:mapper.report(input)) {
@@ -27,11 +29,13 @@ public class InventoryTurnoverQueryService {
                 value=json.readerFor(new TypeReference<Map<String,Object>>(){}).with(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS)
                     .readValue(text(envelope,"payload"));
             } catch(Exception e){throw new IllegalStateException("Invalid turnover query projection",e);}
+            var customerCode=text(value,"customer_code");
+            if(customerCode!=null)value.put("customer_name",names.get(customerCode));
             switch(text(envelope,"section")) {
                 case "ROW" -> rows.add(row(value,request,context));
                 case "TOTAL" -> {total=decimal(value,"total").longValueExact();hasTotal=true;}
                 case "OVERVIEW" -> summary=overview(value,context);
-                case "OWNER" -> owners.add(new OwnerPhysical(text(value,"customer_code"),null,decimal(value,"physical")));
+                case "OWNER" -> owners.add(new OwnerPhysical(text(value,"customer_code"),text(value,"customer_name"),decimal(value,"physical")));
                 case "OWNER_META" -> unknown=count(value,"unknown_count");
                 default -> throw new IllegalStateException("Unknown turnover query section");
             }

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/SelfopOwnerDirectory.java

@@ -0,0 +1,20 @@
+package com.baiyue.lastleg.selfoperated.analytics.query;
+
+import com.baiyue.lastleg.selfoperated.analytics.mapper.SelfopMasterDataMapper;
+import org.springframework.stereotype.Service;
+import java.time.*;
+import java.util.*;
+
+/** Names are read from the last complete master-data batch visible at the query's fixed time. */
+@Service
+public class SelfopOwnerDirectory {
+    private final SelfopMasterDataMapper mapper;
+    public SelfopOwnerDirectory(SelfopMasterDataMapper mapper){this.mapper=mapper;}
+    public Map<String,String> names(String account,Instant asOf) {
+        Long run=mapper.publishedAt(account,LocalDateTime.ofInstant(asOf,ZoneOffset.UTC));
+        if(run==null)return Map.of();
+        var result=new LinkedHashMap<String,String>();
+        for(var owner:mapper.owners(account,run))result.put(owner.customerCode(),owner.customerName()==null||owner.customerName().isBlank()?null:owner.customerName());
+        return Collections.unmodifiableMap(result);
+    }
+}

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressDtos.java

@@ -7,7 +7,7 @@ public final class SelfopEgressDtos {
   boolean continuousEnabled,String sourceAccountFingerprint,boolean grantPresent,String grantFingerprint,
   long intervalMs,int permitMaxSeconds,boolean executionEnabled,int pendingMaxSeconds,
   boolean exclusiveHandoffVerified,boolean priorEgressClosed,boolean inflightDrained,boolean transferredBudgetVerified){}
- public enum Operation {WAREHOUSES,OUTBOUND_QUERY,OUTBOUND_LOCATE,OUTBOUND_DETAIL,PRODUCT_INVENTORY}
+ public enum Operation {WAREHOUSES,OUTBOUND_QUERY,OUTBOUND_LOCATE,OUTBOUND_DETAIL,PRODUCT_INVENTORY,CUSTOMERS,PRODUCTS}
  public record EgressContext(String sourceAccount,String executionPurpose,String rootTaskId,String sessionId,String attemptId,String exclusiveGrantId){}
  @com.fasterxml.jackson.annotation.JsonInclude(com.fasterxml.jackson.annotation.JsonInclude.Include.ALWAYS)
  public record AuthorizationDenial(String errorCode,boolean sampleDone,Instant waitUntil){}

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressService.java

@@ -39,6 +39,9 @@ public class SelfopEgressService {
   SelfopBusinessBudget business=hasBusinessSession(root)?business(root):null;
   if(business!=null){verifyBusinessTask(task,root);String limit=business.limit(now);if(limit!=null){stopBusiness(root,business,limit,now);throw reject(limit);}}
   verify(task,run,root,account,now);
+  boolean masterOperation=operation==Operation.CUSTOMERS||operation==Operation.PRODUCTS;
+  if(masterOperation!=("MASTER_DATA".equals(task.getTaskType()))
+      ||masterOperation&&(!"BUSINESS".equals(task.getExecutionPurpose())||pageSize<1||pageSize>(operation==Operation.CUSTOMERS?1000:100)))throw reject("EGRESS_OPERATION_TASK_MISMATCH");
   checkAccount(account,budget,now);
   SelfopValidationBudget validation=null;
   if("VALIDATION".equals(context.executionPurpose())){

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionService.java

@@ -17,6 +17,7 @@ public class SelfopExecutionService {
     private final SelfopLeaseService leases;
     private final SelfopDiscoveryService discovery;
     private final SelfopInventoryCollector stock;
+    private final SelfopMasterDataCollector masterData;
     private final InventorySnapshotService snapshots;
     private final OutboundFactService facts;
     private final OutboundCandidateBuilder candidates;
@@ -31,9 +32,10 @@ public class SelfopExecutionService {
     public SelfopExecutionService(SelfopAnalyticsProperties config,SelfopTaskService taskService,SelfopLeaseService leases,
         SelfopDiscoveryService discovery,SelfopInventoryCollector stock,InventorySnapshotService snapshots,OutboundFactService facts,
         OutboundCandidateBuilder candidates,SelfopSalesPublisher publisher,SelfopSyncScopeMapper scopes,SelfopSyncTaskMapper tasks,
-        SelfopSyncStageMapper stages,SelfopOutboundOrderMapper orders,SelfopReferenceRecovery recovery,SelfopValidationRunner validation,SelfopSyncRunMapper runs){
+        SelfopSyncStageMapper stages,SelfopOutboundOrderMapper orders,SelfopReferenceRecovery recovery,SelfopValidationRunner validation,SelfopSyncRunMapper runs,SelfopMasterDataCollector masterData){
         this.config=config;this.taskService=taskService;this.leases=leases;this.discovery=discovery;this.stock=stock;this.snapshots=snapshots;
         this.facts=facts;this.candidates=candidates;this.publisher=publisher;this.scopes=scopes;this.tasks=tasks;this.stages=stages;this.orders=orders;this.recovery=recovery;this.validation=validation;this.runs=runs;
+        this.masterData=masterData;
     }
     public void execute(long taskId){
         var task=tasks.selectById(taskId);require(task!=null,"SELFOP_TASK_MISSING");SelfopCollectionPolicy.requireExecution(config,task.getExecutionPurpose());
@@ -63,7 +65,8 @@ public class SelfopExecutionService {
             task=tasks.selectById(taskId);
             if("VALIDATION".equals(task.getExecutionPurpose())){validation.run(task,run,primary);return;}
             if(!primary.account()&&"OUTBOUND".equals(scopes.selectById(task.getScopeId()).getDataFamily()))recovery.refresh(task,held);
-            if(primary.account())discovery.collect(task,run,primary);
+            if("MASTER_DATA".equals(task.getTaskType()))masterData.collect(task,run,primary);
+            else if(primary.account())discovery.collect(task,run,primary);
             else if("INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())){
                 if(stock.collect(task,run,primary)){snapshots.build(task.getScopeId(),run.getId(),owner,primary.fence());snapshots.publish(task.getScopeId(),run.getId(),owner,primary.fence());}
             }else{

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollector.java

@@ -0,0 +1,85 @@
+package com.baiyue.lastleg.selfoperated.analytics.sync;
+
+import com.baomidou.mybatisplus.core.toolkit.IdWorker;
+import com.baiyue.lastleg.selfoperated.analytics.client.LingxingAnalyticsClient;
+import com.baiyue.lastleg.selfoperated.analytics.mapper.*;
+import com.baiyue.lastleg.selfoperated.analytics.model.*;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.support.TransactionTemplate;
+import lombok.Data;
+import java.util.Objects;
+import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*;
+
+/** Account-wide master data uses the same durable task, lease and provider permit as sales. */
+@Service
+public class SelfopMasterDataCollector {
+    private final LingxingAnalyticsClient client;
+    private final SelfopMasterDataMapper data;
+    private final SelfopSyncTaskMapper tasks;
+    private final SelfopSyncRunMapper runs;
+    private final SelfopTaskService taskService;
+    private final SelfopLeaseService leases;
+    private final SelfopAnalyticsProperties config;
+    private final TransactionTemplate tx;
+    public SelfopMasterDataCollector(LingxingAnalyticsClient client,SelfopMasterDataMapper data,SelfopSyncTaskMapper tasks,
+        SelfopSyncRunMapper runs,SelfopTaskService taskService,SelfopLeaseService leases,SelfopAnalyticsProperties config,TransactionTemplate tx) {
+        this.client=client;this.data=data;this.tasks=tasks;this.runs=runs;this.taskService=taskService;this.leases=leases;this.config=config;this.tx=tx;
+    }
+    @Data public static class Manifest {
+        private int nextPage=1,completedPages=0;
+        private Long total;
+        private Integer pages;
+        private long ownerCount,productCount;
+        private boolean ownersComplete;
+    }
+    public void collect(SelfopSyncTask task,SelfopSyncRun run,SelfopLeaseService.Lease lease) {
+        SelfopCollectionPolicy.requireBusiness(task);SelfopCollectionPolicy.requireBusiness(run);
+        require("MASTER_DATA".equals(task.getTaskType())&&lease.account(),"SELFOP_MASTER_TASK_REQUIRED");
+        var manifest=task.getRoutingManifestJson()==null?new Manifest():read(task.getRoutingManifestJson(),Manifest.class);
+        // Re-read priority after at most five pages, without losing the batch or consuming a failed attempt.
+        for(int slice=0;slice<5;slice++) {
+            leases.renew(lease);boolean owners=!manifest.isOwnersComplete();int size=owners?1000:100;
+            var context=SelfopRequestContext.forTask(task,run,config);
+            LingxingAnalyticsClient.Page<?> page=owners?client.customers(context,manifest.nextPage,size):client.products(context,manifest.nextPage,size);
+            validatePage(manifest,page,size);
+            try {tx.executeWithoutResult(status->{
+                var checkpoint=taskService.checkpoint(task,run,lease);
+                for(var value:page.records()) {
+                    if(owners) {var row=(SelfopPayloads.Owner)value;identity(row.customerCode(),64);data.insertOwner(IdWorker.getId(),run.getId(),task.getSourceAccount(),row,utcNow());manifest.ownerCount++;}
+                    else {var row=(SelfopPayloads.Product)value;identity(row.customerCode(),64);identity(row.sku(),128);data.insertProduct(IdWorker.getId(),run.getId(),task.getSourceAccount(),row,utcNow());manifest.productCount++;}
+                }
+                manifest.completedPages++;manifest.nextPage++;
+                boolean complete=page.pages()==0||manifest.nextPage>page.pages();
+                if(complete) {
+                    require((owners?manifest.ownerCount:manifest.productCount)==page.total(),"SELFOP_MASTER_COUNT_MISMATCH");
+                    require((owners?data.ownerCount(run.getId(),task.getSourceAccount()):data.productCount(run.getId(),task.getSourceAccount()))==page.total(),"SELFOP_MASTER_COUNT_MISMATCH");
+                }
+                var t=checkpoint.task();var r=checkpoint.run();
+                SelfopProgressProjection.applyCheckpoint(t,r,owners?"CUSTOMERS":"PRODUCTS","PAGE",Math.min(manifest.nextPage-1,page.pages()),page.pages().longValue(),1);
+                if(complete&&owners){manifest.ownersComplete=true;manifest.nextPage=1;manifest.total=null;manifest.pages=null;}
+                t.setRoutingManifestJson(json(manifest));t.setUpdatedAt(utcNow());
+                if(complete&&!owners) {
+                    r.setStatus("PUBLISHED");r.setLastPhase("PUBLISHED");r.setCoverageStatus("COMPLETE");r.setPageCount(manifest.completedPages);
+                    r.setSourceTotalCount(manifest.ownerCount+manifest.productCount);r.setUniqueObjectCount(manifest.ownerCount+manifest.productCount);
+                    r.setFetchFinishedAtUtc(utcNow());r.setPublishedAtUtc(utcNow());r.setFinishedAtUtc(utcNow());r.setUpdatedAt(utcNow());
+                    t.setStatus("SUCCEEDED");t.setResultPublishedRunId(r.getId());
+                    SelfopProgressProjection.applyCheckpoint(t,r,"PUBLISH","BATCH",1,1L,1);
+                    SelfopProgressProjection.freezeRun(t,r);
+                    tasks.update(null,new com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper<SelfopSyncTask>().eq("id",t.getId()).set("next_retry_at_utc",null));
+                    t.setNextRetryAtUtc(null);
+                }
+                tasks.updateById(t);runs.updateById(r);
+            });} catch(DuplicateKeyException duplicate){throw new IllegalStateException("SELFOP_MASTER_DUPLICATE");}
+            if(!owners&&(page.pages()==0||manifest.nextPage>page.pages()))return;
+        }
+        taskService.yield(task.getId(),run.getId(),lease);
+    }
+    static void validatePage(Manifest m,LingxingAnalyticsClient.Page<?> p,int size) {
+        require(p.current()!=null&&p.current()==m.nextPage&&p.size()!=null&&p.size()==size,"SELFOP_PAGE_MISMATCH");
+        require(p.total()!=null&&p.total()>=0&&p.pages()!=null&&p.pages()==(p.total()+size-1)/size,"SELFOP_PAGE_METADATA_MISSING");
+        require(p.records().size()==Math.min(size,Math.max(0,p.total()-(long)(m.nextPage-1)*size)),"SELFOP_MASTER_COUNT_MISMATCH");
+        if(m.total==null){m.total=p.total();m.pages=p.pages();}
+        require(Objects.equals(m.total,p.total())&&Objects.equals(m.pages,p.pages()),"SELFOP_PAGE_TOTAL_CHANGED");
+    }
+}

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlanner.java

@@ -95,8 +95,22 @@ public class SelfopSchedulePlanner {
                 task.setRequestHash(hash(Arrays.asList(task.getQueryScopeJson(),digest)));taskService.create(task);
             }
             planContinuousInventory(now,today);
+            planMasterData(account,now,today);
         });
     }
+    private void planMasterData(SelfopOutboundDiscoveryScope account,LocalDateTime now,LocalDate today) {
+        // One durable daily slot; unfinished work is resumed rather than replaced by a new day's scan.
+        if(tasks.selectCount(new QueryWrapper<SelfopSyncTask>().eq("source_account",config.getSourceAccount())
+            .eq("task_type","MASTER_DATA").in("status","READY","RUNNING","RETRY_WAIT"))>0)return;
+        var definition=new TreeMap<String,Object>();definition.put("dataFamily","MASTER_DATA");definition.put("sourceAccount",config.getSourceAccount());
+        String digest=hash(definition),key="continuous:master:"+today+":"+digest.substring(0,12);
+        if(tasks.selectCount(new QueryWrapper<SelfopSyncTask>().eq("source_account",config.getSourceAccount()).eq("request_key",key))>0)return;
+        var task=new SelfopSyncTask();task.setSourceAccount(config.getSourceAccount());task.setExecutionPurpose("BUSINESS");
+        task.setTaskRole("ACCOUNT_DISCOVERY");task.setTaskType("MASTER_DATA");task.setDiscoveryScopeId(account.getId());
+        task.setRequestedBy(SelfopCollectionPolicy.CONTINUOUS_PLANNER);task.setScopeHash(digest);task.setRequestKey(key);
+        task.setPlannedAtUtc(today.atStartOfDay().minusHours(8));task.setQueryScopeJson(json(definition));
+        task.setRequestHash(hash(Arrays.asList(task.getQueryScopeJson(),digest)));taskService.create(task);
+    }
     private void planContinuousInventory(LocalDateTime now,LocalDate today){
         var inventory=scopes.selectList(new QueryWrapper<SelfopSyncScope>().eq("source_account",config.getSourceAccount())
             .eq("data_family","INVENTORY").in("warehouse_code",config.getWarehouses().keySet()));

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java

@@ -151,7 +151,7 @@ public class SelfopTaskService {
         require(!"SAMPLE_DONE".equals(task.getStatus()),"SELFOP_VALIDATION_TERMINATED");
         if(lease.account()) {var account=locks.lockDiscovery(lease.scopeId());account.setActiveTaskId(taskId);account.setUpdatedAt(utcNow());accounts.updateById(account);}
         var now=utcNow();var active=runs.selectOne(new QueryWrapper<SelfopSyncRun>().eq("task_id",taskId).eq("status","FETCHING").orderByDesc("id").last("LIMIT 1"));
-        if(task.getScopeId()!=null && "INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())) {
+        if("MASTER_DATA".equals(task.getTaskType()) || task.getScopeId()!=null && "INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())) {
             var yielded=runs.selectOne(new QueryWrapper<SelfopSyncRun>().eq("task_id",taskId).eq("status","YIELDED").orderByDesc("id").last("LIMIT 1"));
             if(yielded!=null){yielded.setStatus("FETCHING");yielded.setFinishedAtUtc(null);yielded.setProgressSnapshotJson(null);yielded.setFenceToken(lease.fence());yielded.setUpdatedAt(now);
                 runs.update(null,new UpdateWrapper<SelfopSyncRun>().eq("id",yielded.getId()).set("status","FETCHING").set("fence_token",lease.fence()).set("finished_at_utc",null).set("progress_snapshot_json",null).set("updated_at",now));
@@ -160,7 +160,7 @@ public class SelfopTaskService {
         if(active!=null){require(!Objects.equals(active.getFenceToken(),lease.fence()),"SELFOP_TASK_ALREADY_RUNNING");
             SelfopProgressProjection.freezeRun(task,active);
             active.setStatus("ABORTED");active.setLastPhase("FETCHING");active.setFinishedAtUtc(now);active.setErrorCode("LEASE_EXPIRED");active.setErrorMessage("执行租约失效");active.setUpdatedAt(now);runs.updateById(active);}
-        if(task.getScopeId()!=null&&"INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())) {
+        if("MASTER_DATA".equals(task.getTaskType()) || task.getScopeId()!=null&&"INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())) {
             // Every genuinely new inventory attempt starts at page one, including a crash that never called fail().
             task.setRoutingManifestJson(null);
             tasks.update(null,new UpdateWrapper<SelfopSyncTask>().eq("id",taskId).set("routing_manifest_json",null));
@@ -180,7 +180,7 @@ public class SelfopTaskService {
                 if(sales!=null)sales=locks.lockScope(sales.getId());
                 run.setReferenceSalesRunId(sales==null?null:sales.getPublishedRunId());}}
         runs.insert(run);task.setAttemptCount(run.getAttemptNo());task.setStatus("RUNNING");task.setUpdatedAt(now);
-        String initialPhase=lease.account()?"DIRECTORY":("INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())?"INVENTORY":"MERGE");
+        String initialPhase="MASTER_DATA".equals(task.getTaskType())?"CUSTOMERS":lease.account()?"DIRECTORY":("INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())?"INVENTORY":"MERGE");
         SelfopProgressProjection.applyCheckpoint(task,run,initialPhase,"PAGE",0,null,1);tasks.updateById(task);return run;
     }
     @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
@@ -202,7 +202,8 @@ public class SelfopTaskService {
         if(task.getBusinessRootTaskId()!=null&&continuousBusinessRoot(tasks.selectById(task.getBusinessRootTaskId()))) {
             // A completed response with an unstable inventory scan can be read again from page one.
             // Authentication, rate limits, uncertain transport results and other families retain their stop policy.
-            retryable=retryable&&inventory&&Set.of("SELFOP_STOCK_DUPLICATE","SELFOP_STOCK_COUNT_MISMATCH","SELFOP_PAGE_TOTAL_CHANGED").contains(code);
+            retryable=retryable&&(inventory&&Set.of("SELFOP_STOCK_DUPLICATE","SELFOP_STOCK_COUNT_MISMATCH","SELFOP_PAGE_TOTAL_CHANGED").contains(code)
+                ||"MASTER_DATA".equals(task.getTaskType())&&Set.of("SELFOP_MASTER_DUPLICATE","SELFOP_MASTER_COUNT_MISMATCH","SELFOP_PAGE_TOTAL_CHANGED").contains(code));
         }
         task.setStatus(retryable&&failures<config.getMaxAttempts()?"RETRY_WAIT":"DEAD");
         if("APPLYING".equals(task.getApplyState())) task.setApplyState("BLOCKED");