From d311d9dc8ddcd0832ba49d071da07264129f2501 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Wed, 14 Oct 2026 16:15:39 +0800 Subject: [PATCH] feat(selfop): sync versioned owner and product directories --- deploy/sql/README.md | 6 +- deploy/sql/海外区域统一增量-MySQL8.sql | 37 ++++++- .../LingxingInternalController.java | 18 ++++ .../dto/lingxing/LingxingEgressDtos.java | 2 +- .../dto/lingxing/LingxingWmsDtos.java | 5 + .../provider/lingxing/LingxingWmsClient.java | 18 ++++ .../resources/openapi/lingxing-wms-v1.yaml | 98 ++++++++++++++++++- .../lingxing/LingxingWmsClientTest.java | 17 ++++ .../client/LingxingAnalyticsClient.java | 2 + .../client/RestLingxingAnalyticsClient.java | 15 +++ .../PlatformAnalyticsController.java | 6 ++ .../mapper/SelfopMasterDataMapper.java | 33 +++++++ .../model/SelfopCollectionPolicy.java | 9 ++ .../analytics/model/SelfopPayloads.java | 4 + .../model/SelfopProgressProjection.java | 2 +- .../query/AnalyticsQueryService.java | 36 ++++--- .../analytics/query/AnalyticsResponses.java | 2 +- .../analytics/query/CollectionReadMapper.java | 2 +- .../query/DashboardQueryService.java | 18 ++-- .../analytics/query/DashboardResponses.java | 2 +- .../query/InventoryTurnoverContext.java | 10 +- .../query/InventoryTurnoverPolicy.java | 2 +- .../query/InventoryTurnoverQueryService.java | 8 +- .../analytics/query/SelfopOwnerDirectory.java | 20 ++++ .../analytics/request/SelfopEgressDtos.java | 2 +- .../request/SelfopEgressService.java | 3 + .../sync/SelfopExecutionService.java | 7 +- .../sync/SelfopMasterDataCollector.java | 85 ++++++++++++++++ .../analytics/sync/SelfopSchedulePlanner.java | 14 +++ .../analytics/sync/SelfopTaskService.java | 9 +- .../src/main/resources/db/lastleg.sql | 34 +++++++ .../main/resources/db/selfop-master-data.sql | 33 +++++++ .../self-operated-analytics-egress-v1.yaml | 2 +- .../openapi/self-operated-analytics-v1.yaml | 35 ++++++- .../RestLingxingAnalyticsClientTest.java | 14 +++ .../PlatformAnalyticsControllerTest.java | 5 +- .../mapper/SelfopMasterDataMapperTest.java | 60 ++++++++++++ .../query/AnalyticsQueryServiceTest.java | 11 ++- .../query/DashboardQueryServiceTest.java | 4 +- .../query/InventoryTurnoverContextTest.java | 6 +- .../InventoryTurnoverQueryServiceTest.java | 21 ++++ .../query/SelfopOwnerDirectoryTest.java | 17 ++++ .../request/SelfopEgressServiceTest.java | 19 ++++ .../sync/SelfopContinuousDueSqlTest.java | 6 ++ .../sync/SelfopExecutionServiceTest.java | 4 +- .../sync/SelfopMasterDataCollectorTest.java | 76 ++++++++++++++ .../sync/SelfopSchedulePlannerTest.java | 30 ++++-- 47 files changed, 803 insertions(+), 66 deletions(-) create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopMasterDataMapper.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/SelfopOwnerDirectory.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollector.java create mode 100644 lastleg-service/src/main/resources/db/selfop-master-data.sql create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopMasterDataMapperTest.java create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverQueryServiceTest.java create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/SelfopOwnerDirectoryTest.java create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollectorTest.java diff --git a/deploy/sql/README.md b/deploy/sql/README.md index d860394dd..c2464d426 100644 --- a/deploy/sql/README.md +++ b/deploy/sql/README.md @@ -28,6 +28,8 @@ 平台自营领星统计十张 `ll_selfop_*` 表直接同步 Lastleg 源基线及海外统一空库基线；时间使用 UTC 存储及已确认北京时间业务日，数量使用 DECIMAL 保留 PCS 精度，维持独立发布版本。包含账号共享预算/冷却、验证根用途与硬预算、TEST BUSINESS 根会话持久预算、稳定任务号、阶段进度/心跳和旧版替代时间；任务的 `business_root_task_id` 标记共享根，只有根保存 `business_budget_json`，生产任务不使用该测试预算。库存 run 包含每日代表日期、选择规则及账号/仓库/日期唯一约束，代表库存与 profile 不适用销量副本淘汰。IAM 权限源为 `iam/src/main/resources/db/selfoperated-analytics-permissions.sql`，自营业务下经营总览、销量分析、库存分析、数据采集的菜单资源与独立读写权限同步国内统一基线，仅平台管理员默认授权。执行和部署状态只维护于工作区测试环境主契约；真实采集模式默认 OFF。 +客户/商品资料基线为 `lastleg-service/src/main/resources/db/selfop-master-data.sql`，仅新增 `ll_selfop_owner`、`ll_selfop_product` 两表，并同步在 Lastleg 和海外统一空库基线。按来源账号、货主编码、SKU 和完整资料版本关联，不创建平台客户、收费对象或独立映射表；未知实测尺寸保持 NULL。对 TEST 已有库只审阅并授权这两张新表的 CREATE，不能重放统一初始化文件，也不重置采集数据。 + ## 唯一入口与区域边界 两个统一 SQL 文件仍只用于空库初始化，名称中的“增量”不表示可向已有业务库重放。已有生产库的受控结构补齐单独放在 `migrations/`，只允许明确审阅过的增量变更；不提供存量数据回填、清理或兼容入口。 @@ -35,7 +37,7 @@ | 区域 | 完整入口 | 服务 schema | 表数 | |---|---|---|---:| | 国内 | `头程数据库统一增量-MySQL8.sql` | `baiyue_auth`、`baiyue_iam`、`baiyue_order`、`baiyue_warehouse`、`baiyue_firstleg`、`baiyue_file`、`baiyue_audit`、`baiyue_finance` | 226 | -| 海外 | `海外区域统一增量-MySQL8.sql` | `baiyue_lastleg`、`baiyue_lastlag_warehouse` | 195 | +| 海外 | `海外区域统一增量-MySQL8.sql` | `baiyue_lastleg`、`baiyue_lastlag_warehouse` | 199 | - 所有服务位于同一测试环境；两份文件只分组组织服务 schema，不代表两个部署区域。`baiyue_lastlag_warehouse` 保留当前正式的 `lastlag` 拼写。 - 两份文件必须在同一测试数据库实例中完成初始化；业务服务仍只访问本服务 schema，经内部接口访问 IAM/Warehouse。已移除事件桥两个 schema 的创建，不对已有数据库执行 DROP。 @@ -57,7 +59,7 @@ 5. DDL 隐式提交，不可整体回滚。失败后停止并保留错误；经另行授权准备空库后从头执行，不针对半初始化库追加迁移或从失败行续跑。 6. Firstleg 测量项与拆分项互为外键：建空表期间暂设会话 `foreign_key_checks=0`，随后立即恢复 `1`，再插入种子。失败时关闭连接。两边外键完整保留在 CREATE TABLE 内。 -成功时只在最末输出一行 `initialization_status=PASS`，并带 `region=CN/OVERSEAS`、`initialized_tables=226/194`。必须同时确认客户端退出码 0；忽略错误后出现的 PASS 不算成功。 +成功时只在最末输出一行 `initialization_status=PASS`，并带 `region=CN/OVERSEAS`、`initialized_tables=226/199`。必须同时确认客户端退出码 0；忽略错误后出现的 PASS 不算成功。 ## 平台管理员密码 diff --git a/deploy/sql/海外区域统一增量-MySQL8.sql b/deploy/sql/海外区域统一增量-MySQL8.sql index 616512ab8..f97c1dbad 100644 --- a/deploy/sql/海外区域统一增量-MySQL8.sql +++ b/deploy/sql/海外区域统一增量-MySQL8.sql @@ -4235,7 +4235,6 @@ CREATE TABLE ll_selfop_sku_profile ( KEY idx_selfop_profile_evidence (profile_run_id,earliest_evidence_at_utc,customer_code,sku_code) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='随发布版本固定的 SKU 维度与观察依据'; -SELECT 'PASS' AS initialization_status, 'OVERSEAS' AS region, 194 AS initialized_tables; CREATE TABLE IF NOT EXISTS ll_appointment_impact ( @@ -4321,3 +4320,39 @@ CREATE TABLE IF NOT EXISTS ll_cargo_discard_photo ( UNIQUE KEY uk_cargo_discard_photo_file (file_id), KEY idx_cargo_discard_photo_application (customer_id, application_id, pallet_no) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE ll_selfop_owner ( + id BIGINT NOT NULL COMMENT '主键', + run_id BIGINT NOT NULL COMMENT '完整资料采集版本', + source_account VARCHAR(64) COLLATE utf8mb4_bin NOT NULL COMMENT '来源账号', + customer_code VARCHAR(64) COLLATE utf8mb4_bin NOT NULL COMMENT '领星货主编码', + customer_name VARCHAR(255) NULL COMMENT '领星货主名称', + source_status VARCHAR(32) NULL COMMENT '来源启停状态', + observed_at_utc DATETIME(6) NOT NULL COMMENT '资料观测时间', + PRIMARY KEY (id), + UNIQUE KEY uk_selfop_owner_version (run_id,source_account,customer_code), + KEY idx_selfop_owner_lookup (source_account,customer_code,run_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='自营领星货主资料版本'; + +CREATE TABLE ll_selfop_product ( + id BIGINT NOT NULL COMMENT '主键', + run_id BIGINT NOT NULL COMMENT '完整资料采集版本', + source_account VARCHAR(64) COLLATE utf8mb4_bin NOT NULL COMMENT '来源账号', + customer_code VARCHAR(64) COLLATE utf8mb4_bin NOT NULL COMMENT '领星货主编码', + sku_code VARCHAR(128) COLLATE utf8mb4_bin NOT NULL COMMENT '货主商品编码', + product_name VARCHAR(255) NULL COMMENT '商品名称', + declared_length DECIMAL(24,6) NULL COMMENT '商品资料长度', + declared_width DECIMAL(24,6) NULL COMMENT '商品资料宽度', + declared_height DECIMAL(24,6) NULL COMMENT '商品资料高度', + declared_size_unit VARCHAR(32) NULL COMMENT '商品资料尺寸单位', + measured_length DECIMAL(24,6) NULL COMMENT '仓库实测长度', + measured_width DECIMAL(24,6) NULL COMMENT '仓库实测宽度', + measured_height DECIMAL(24,6) NULL COMMENT '仓库实测高度', + measured_size_unit VARCHAR(32) NULL COMMENT '仓库实测尺寸单位', + observed_at_utc DATETIME(6) NOT NULL COMMENT '资料观测时间', + PRIMARY KEY (id), + UNIQUE KEY uk_selfop_product_version (run_id,source_account,customer_code,sku_code), + KEY idx_selfop_product_lookup (source_account,customer_code,sku_code,run_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='自营领星商品资料版本'; + +SELECT 'PASS' AS initialization_status, 'OVERSEAS' AS region, 199 AS initialized_tables; diff --git a/integration-service/src/main/java/com/baiyue/integration/controller/LingxingInternalController.java b/integration-service/src/main/java/com/baiyue/integration/controller/LingxingInternalController.java index 7458395e3..ec969e957 100644 --- a/integration-service/src/main/java/com/baiyue/integration/controller/LingxingInternalController.java +++ b/integration-service/src/main/java/com/baiyue/integration/controller/LingxingInternalController.java @@ -88,6 +88,24 @@ public class LingxingInternalController { return ApiResponse.success(client.queryInventory(request, context(context))); } + @PostMapping("/customers/query") + public ApiResponse> customers( + @RequestHeader(value="X-Internal-Service-Token",required=false) String token, + @RequestHeader(value="X-Lingxing-Egress-Context",required=false) String context, + @Valid @RequestBody WarehouseQueryRequest request) { + tokenGuard.verify(token); + return ApiResponse.success(client.queryCustomers(request,context(context))); + } + + @PostMapping("/products/query") + public ApiResponse> products( + @RequestHeader(value="X-Internal-Service-Token",required=false) String token, + @RequestHeader(value="X-Lingxing-Egress-Context",required=false) String context, + @Valid @RequestBody ProductQueryRequest request) { + tokenGuard.verify(token); + return ApiResponse.success(client.queryProducts(request,context(context))); + } + private EgressContext context(String header) { if (header == null || header.isBlank() || header.length() > 2048 || !header.chars().allMatch(c -> c >= 32 && c <= 126)) throw invalidContext(); diff --git a/integration-service/src/main/java/com/baiyue/integration/dto/lingxing/LingxingEgressDtos.java b/integration-service/src/main/java/com/baiyue/integration/dto/lingxing/LingxingEgressDtos.java index 5d02a6808..e9749ff69 100644 --- a/integration-service/src/main/java/com/baiyue/integration/dto/lingxing/LingxingEgressDtos.java +++ b/integration-service/src/main/java/com/baiyue/integration/dto/lingxing/LingxingEgressDtos.java @@ -6,7 +6,7 @@ import java.time.Instant; public final class LingxingEgressDtos { private LingxingEgressDtos() { } public enum CollectionMode { OFF, VALIDATION, BUSINESS } - 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, CollectionMode executionPurpose, String rootTaskId, String sessionId, String attemptId, String exclusiveGrantId) { public boolean wellFormed() { 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 1e43243bb..2a0eefeae 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 @@ -19,6 +19,11 @@ public final class LingxingWmsDtos { @Min(1) int current, @Min(1) @Max(100) int size) { } public record WarehouseQueryRequest(@Min(1) int current, @Min(1) @Max(1000) int size) { } + public record ProductQueryRequest(@Min(1) int current, @Min(1) @Max(100) int size) { } + public record Customer(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 LocateRequest(@NotBlank @Size(max=128) String deliveryNo, @Min(1) int current, @Min(1) @Max(1000) int size) { } public record DetailRequest(@NotBlank @Size(max=64) String warehouseCode, 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 ce9d64d6b..0281cd1af 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 @@ -133,6 +133,24 @@ public final class LingxingWmsClient { optionalText(data, "outboundTime"), lines, 200, clock.instant(), metadata); } + public QueryPage queryCustomers(WarehouseQueryRequest request, EgressContext context) { + if (request == null) throw badRequest(); + validatePagination(request.current(), request.size(), 1000); + return execute("/openapi/v2/customer/list", pagination(request.current(), request.size()), context, Operation.CUSTOMERS, + (data, metadata) -> mappedPage(data, request.current(), request.size(), "current", "size", false, metadata, + row -> new Customer(requiredText(row,"customerCode",64), optionalText(row,"customerName"), optionalText(row,"status")))); + } + + public QueryPage queryProducts(ProductQueryRequest request, EgressContext context) { + if (request == null) throw badRequest(); + validatePagination(request.current(), request.size(), 100); + return execute("/openapi/v2/product/page", pagination(request.current(), request.size()), context, Operation.PRODUCTS, + (data, metadata) -> mappedPage(data, request.current(), request.size(), "current", "size", false, metadata, + row -> new Product(requiredText(row,"customerCode",64), requiredText(row,"sku",128), optionalText(row,"productName"), + quantity(row.get("length")),quantity(row.get("width")),quantity(row.get("height")),optionalText(row,"sizeUnit"), + quantity(row.get("wmsLength")),quantity(row.get("wmsWidth")),quantity(row.get("wmsHeight")),optionalText(row,"wmsSizeUnit")))); + } + public QueryPage queryInventory(InventoryQueryRequest request, EgressContext context) { if (request == null) throw badRequest(); requireInput(request.warehouseCode(), 64); 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 9ebf851be..7c327a5a1 100644 --- a/integration-service/src/main/resources/openapi/lingxing-wms-v1.yaml +++ b/integration-service/src/main/resources/openapi/lingxing-wms-v1.yaml @@ -51,6 +51,40 @@ x-lastleg-egress-coordination: security: - InternalServiceToken: [] paths: + /internal/integrations/lingxing/wms/customers/query: + post: + summary: Query one page of customers using the existing account egress permit + description: Calls /openapi/v2/customer/list. BUSINESS MASTER_DATA tasks only; no retries or separate egress path. Dimensions and units remain separate; blanks are null. + parameters: + - {$ref: '#/components/parameters/LingxingEgressContext'} + requestBody: + required: true + content: + application/json: + schema: {$ref: '#/components/schemas/WarehouseQueryRequest'} + responses: + '200': + description: Standard envelope with one page or sanitized ProviderFailure. + content: + application/json: + schema: {$ref: '#/components/schemas/CustomerResponse'} + /internal/integrations/lingxing/wms/products/query: + post: + summary: Query one page of products using the existing account egress permit + description: Calls /openapi/v2/product/page. BUSINESS MASTER_DATA tasks only; no retries or separate egress path. Dimensions and units remain separate; blanks are null. + parameters: + - {$ref: '#/components/parameters/LingxingEgressContext'} + requestBody: + required: true + content: + application/json: + schema: {$ref: '#/components/schemas/ProductQueryRequest'} + responses: + '200': + description: Standard envelope with one page or sanitized ProviderFailure. + content: + application/json: + schema: {$ref: '#/components/schemas/ProductResponse'} /internal/integrations/lingxing/wms/egress/runtime: get: operationId: lingxingEgressRuntime @@ -175,6 +209,68 @@ components: in: header name: X-Internal-Service-Token schemas: + CustomerPage: + allOf: + - {$ref: '#/components/schemas/Pagination'} + - type: object + required: [records] + properties: + records: {type: array, items: {$ref: '#/components/schemas/Customer'}} + CustomerResponse: + allOf: + - {$ref: '#/components/schemas/Envelope'} + - type: object + properties: + data: + nullable: true + oneOf: + - {$ref: '#/components/schemas/CustomerPage'} + - {$ref: '#/components/schemas/ProviderFailure'} + ProductPage: + allOf: + - {$ref: '#/components/schemas/Pagination'} + - type: object + required: [records] + properties: + records: {type: array, items: {$ref: '#/components/schemas/Product'}} + ProductResponse: + allOf: + - {$ref: '#/components/schemas/Envelope'} + - type: object + properties: + data: + nullable: true + oneOf: + - {$ref: '#/components/schemas/ProductPage'} + - {$ref: '#/components/schemas/ProviderFailure'} + ProductQueryRequest: + type: object + required: [current, size] + properties: + current: {type: integer, minimum: 1} + size: {type: integer, minimum: 1, maximum: 100} + Customer: + type: object + required: [customerCode] + properties: + customerCode: {type: string} + customerName: {type: string, nullable: true} + status: {type: string, nullable: true} + Product: + type: object + required: [customerCode, sku] + properties: + customerCode: {type: string} + sku: {type: string} + productName: {type: string, nullable: true} + length: {type: number, nullable: true} + width: {type: number, nullable: true} + height: {type: number, nullable: true} + wmsLength: {type: number, nullable: true} + wmsWidth: {type: number, nullable: true} + wmsHeight: {type: number, nullable: true} + sizeUnit: {type: string, nullable: true} + wmsSizeUnit: {type: string, nullable: true} EgressRuntimeSnapshot: type: object additionalProperties: false @@ -219,7 +315,7 @@ components: {"current":1,"size":2,"whCode":"WH-DEMO","isHideInventory":0} EgressOperation: type: string - enum: [WAREHOUSES, OUTBOUND_QUERY, OUTBOUND_LOCATE, OUTBOUND_DETAIL, PRODUCT_INVENTORY] + enum: [WAREHOUSES, OUTBOUND_QUERY, OUTBOUND_LOCATE, OUTBOUND_DETAIL, PRODUCT_INVENTORY, CUSTOMERS, PRODUCTS] EgressMetadata: type: object required: [attemptId, actualHttpAttempts, accountPaused] 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 2ba24e33a..02da15c86 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 @@ -22,6 +22,23 @@ import static org.junit.jupiter.api.Assertions.assertThrows; /** Synthetic loopback transport only; no supplier or application services. */ class LingxingWmsClientTest { + @Test void customerDirectoryKeepsNamesAndUsesAccountGate() { + response.set(page("current","size",1,1000,1,1,"{\"customerCode\":\"C\",\"customerName\":\"示例货主\",\"status\":\"0\",\"telephoneNo\":\"not-forwarded\"}")); + var result=client().queryCustomers(new WarehouseQueryRequest(1,1000),context()); + assertThat(path.get()).isEqualTo("/openapi/v2/customer/list"); + assertThat(result.records()).containsExactly(new Customer("C","示例货主","0")); + assertThat(completion.get().operation()).isEqualTo(Operation.CUSTOMERS); + } + @Test void productDirectoryRetainsDeclaredDimensionsAndUnknownMeasurements() { + response.set(page("current","size",1,100,1,1,"{\"customerCode\":\"C\",\"sku\":\"SKU\",\"productName\":\"商品\",\"length\":5,\"width\":11,\"height\":5,\"sizeUnit\":\"cm\",\"wmsLength\":\"\",\"wmsWidth\":\"\",\"wmsHeight\":\"\",\"wmsSizeUnit\":\"\"}")); + var result=client().queryProducts(new ProductQueryRequest(1,100),context()); + assertThat(result.records().get(0).length()).isEqualByComparingTo("5"); + assertThat(result.records().get(0).wmsLength()).isNull(); + assertThat(result.records().get(0).sku()).isEqualTo("SKU"); + assertThat(completion.get().operation()).isEqualTo(Operation.PRODUCTS); + assertThatThrownBy(()->client().queryProducts(new ProductQueryRequest(1,101),context())).isInstanceOf(LingxingWmsException.class); + assertThat(calls.get()).isEqualTo(1); + } private static final Instant NOW = Instant.parse("2026-10-12T00:00:00Z"); private static final String PRIVATE = "private-secret@example.test"; private static final String ORDER = """ 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 a492f4c2b..f36a4dfcb 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 @@ -13,6 +13,8 @@ public interface LingxingAnalyticsClient { Page locate(SelfopRequestContext context, String deliveryNo, int current, int size); Order detail(SelfopRequestContext context, String warehouseCode, String customerCode, String deliveryNo); Page inventory(SelfopRequestContext context, String warehouseCode, int current, int size); + Page customers(SelfopRequestContext context, int current, int size); + Page products(SelfopRequestContext context, int current, int size); record Page(List records, Integer current, Integer size, Long total, Integer pages) { public Page { records = List.copyOf(records); } 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 69435ef77..5c343f238 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 @@ -109,6 +109,19 @@ public class RestLingxingAnalyticsClient implements LingxingAnalyticsClient { }); } + @Override public Page 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 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 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; diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/controller/PlatformAnalyticsController.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/controller/PlatformAnalyticsController.java index fdcf2fdb3..bb913cc6c 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/controller/PlatformAnalyticsController.java +++ b/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> ownerDirectory( + @AuthenticationPrincipal AuthenticatedUser user,@ModelAttribute AnalyticsQuery query){ + requirePlatform(user);return ApiResponse.success(service.ownerDirectory(query)); + } @GetMapping("/owner-overview") @Operation(summary="本仓一个货主完整候选的30日经营概览") public ApiResponse ownerOverview( diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopMasterDataMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopMasterDataMapper.java new file mode 100644 index 000000000..c316c3e86 --- /dev/null +++ b/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 owners(@Param("account") String account,@Param("run") long run); +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopCollectionPolicy.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopCollectionPolicy.java index 1998b865c..5ae1f2270 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopCollectionPolicy.java +++ b/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); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopPayloads.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopPayloads.java index 2bf453e96..0140d0965 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopPayloads.java +++ b/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 items) { } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopProgressProjection.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopProgressProjection.java index bd72168e0..b09f14a31 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopProgressProjection.java +++ b/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 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); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryService.java index c32906d0d..4c6f9b1f3 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryService.java +++ b/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 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 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 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 statistics=mapper.rankingStats(p); long total=statistics.isEmpty()?0:statistics.get(0).total(); - List rows=q.offset()>=total?List.of():mapper.page(p).stream().map(r->metric(s,r,kind)).toList(); + var names=ownerNames(s);List rows=q.offset()>=total?List.of():mapper.page(p).stream().map(r->metric(s,r,kind,names.get(r.customerCode()))).toList(); Map 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 ownerNames(Snapshot s){return ownerDirectory.names(s.q.sourceAccount(),s.context.queryAsOf());} + public List 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 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()); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsResponses.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsResponses.java index acada1b11..87508f48b 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsResponses.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsResponses.java @@ -17,7 +17,7 @@ public final class AnalyticsResponses { public record Page(Context context,List records,int current,int size,long total,long pages,Map excludedReasons){} public record SalesSummary(String sourceAccount,String warehouseCode,String customerCode,String sku,String productName, String quantityUnit,BigDecimal outboundQuantity,BigDecimal observedQuantity,String dataStatus,List 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 reasonCodes,String quantityUnit,Instant coveredFrom,Instant coveredUntil){} public record Trend(Context context,SalesSummary summary,List points){} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java index 6981a31e2..6234e6454 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java +++ b/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 (" diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardQueryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardQueryService.java index 7786c7fd2..b92513ca9 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardQueryService.java +++ b/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 s,List rows,boolean available,LocalDate from,LocalDate until,AnalyticsResponses.Context c){ + private static DashboardResponses.Ranking ranking(String kind,Map s,List rows,boolean available,LocalDate from,LocalDate until,AnalyticsResponses.Context c,Map 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 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.of(); + divide("HOT".equals(kind)?r.ownerOutbound():r.outboundQuantity30d(),windowDays),"HOT".equals(kind)?null:r.coverageDays(),names.get(r.customerCode()))).toList():List.of(); long unknown=count(s,prefix+"_unknown");String status=!available?"UNKNOWN":unknown>0?"PARTIAL":"COMPLETE"; List 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"); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardResponses.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardResponses.java index 09f01b104..52edf148a 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardResponses.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardResponses.java @@ -20,7 +20,7 @@ public final class DashboardResponses { List buckets,String dataStatus,List reasonCodes,Map 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 items,String dataStatus,List reasonCodes){} public record Dashboard(Context context,Kpis kpis,SalesTrend salesTrend,Distribution coverageDistribution, diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContext.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContext.java index 5f3abca3b..00e24df04 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContext.java +++ b/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(); for(var sample:mapper.samples(q.sourceAccount(),q.warehouseCode(),inventoryHash,utc(asOf),end.minusDays(Math.max(30,q.days())-1),end.plusDays(1))) { diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicy.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicy.java index a3075b78e..07565b737 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicy.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverPolicy.java @@ -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), diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverQueryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverQueryService.java index 40a5e2c22..8466b59ed 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverQueryService.java +++ b/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 rows=new ArrayList<>();List owners=new ArrayList<>();int unknown=0; for(var part:mapper.report(input)) { @@ -27,11 +29,13 @@ public class InventoryTurnoverQueryService { value=json.readerFor(new TypeReference>(){}).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"); } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/SelfopOwnerDirectory.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/SelfopOwnerDirectory.java new file mode 100644 index 000000000..b051775a3 --- /dev/null +++ b/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 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(); + for(var owner:mapper.owners(account,run))result.put(owner.customerCode(),owner.customerName()==null||owner.customerName().isBlank()?null:owner.customerName()); + return Collections.unmodifiableMap(result); + } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressDtos.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressDtos.java index a48931710..34f6b706e 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressDtos.java +++ b/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){} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressService.java index 955a10473..144d519fc 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressService.java +++ b/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())){ diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionService.java index d7855cf73..b29965ba9 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionService.java +++ b/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{ diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollector.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollector.java new file mode 100644 index 000000000..31a8403c5 --- /dev/null +++ b/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().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"); + } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlanner.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlanner.java index 3503cc907..131bb315e 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlanner.java +++ b/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().eq("source_account",config.getSourceAccount()) + .eq("task_type","MASTER_DATA").in("status","READY","RUNNING","RETRY_WAIT"))>0)return; + var definition=new TreeMap();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().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().eq("source_account",config.getSourceAccount()) .eq("data_family","INVENTORY").in("warehouse_code",config.getWarehouses().keySet())); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java index 111b0fbb5..fe2ac214c 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java +++ b/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().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().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().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().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 client.warehouses(context("synthetic-account"), 1, 100)).hasMessage("EXECUTION_DISABLED"); enable(); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/controller/PlatformAnalyticsControllerTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/controller/PlatformAnalyticsControllerTest.java index bffb5c563..c0529f424 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/controller/PlatformAnalyticsControllerTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/controller/PlatformAnalyticsControllerTest.java @@ -26,6 +26,7 @@ class PlatformAnalyticsControllerTest { var controller=new PlatformAnalyticsController(service,collection,mock(DashboardQueryService.class)); var user=new AuthenticatedUser(1L,2L,2L,1L,"demo","CLIENT","WEB",Set.of(PlatformAnalyticsController.READ_PERMISSION)); assertThrows(ResponseStatusException.class,()->controller.inventoryCoverage(user,null,null)); + assertThrows(ResponseStatusException.class,()->controller.ownerDirectory(user,null)); assertThrows(ResponseStatusException.class,()->controller.syncTaskDetail(user,"LXCOL-DEMO",null)); verifyNoInteractions(service,collection); } @@ -34,6 +35,7 @@ class PlatformAnalyticsControllerTest { var controller=new PlatformAnalyticsController(service,collection,mock(DashboardQueryService.class)); var user=new AuthenticatedUser(1L,0L,0L,1L,"demo","PLATFORM","WEB",Set.of("platform:self-operated-analytics:sync")); assertThrows(ResponseStatusException.class,()->controller.hotSkus(user,null)); + assertThrows(ResponseStatusException.class,()->controller.ownerDirectory(user,null)); assertThrows(ResponseStatusException.class,()->controller.syncOverview(user,null)); assertThrows(ResponseStatusException.class,()->controller.slowSkus(null,null)); verifyNoInteractions(service,collection); @@ -42,7 +44,8 @@ class PlatformAnalyticsControllerTest { var service=mock(AnalyticsQueryService.class);var collection=mock(CollectionQueryService.class); var c=new PlatformAnalyticsController(service,collection,mock(DashboardQueryService.class));var user=platform(); c.salesSummary(user,null,com.baiyue.lastleg.selfoperated.analytics.query.SalesSort.QUANTITY_DESC);c.dailySales(user,null);c.hotSkus(user,null);c.slowSkus(user,null);c.inventoryCoverage(user,null,null); - c.owners(user,null,null);c.ownerOverview(user,null,null);c.syncOverview(user,null);c.syncTasks(user,null);c.syncTaskDetail(user,"LXCOL-DEMO",null);c.syncRuns(user,null); + c.ownerDirectory(user,null);c.owners(user,null,null);c.ownerOverview(user,null,null);c.syncOverview(user,null);c.syncTasks(user,null);c.syncTaskDetail(user,"LXCOL-DEMO",null);c.syncRuns(user,null); + verify(service).ownerDirectory(null); verify(service).salesSummary(null,com.baiyue.lastleg.selfoperated.analytics.query.SalesSort.QUANTITY_DESC);verify(service).dailySales(null);verify(service).hotSkus(null); verify(service).slowSkus(null);verify(service).inventoryCoverage(null,null); verify(service).owners(null,null);verify(service).ownerOverview(null,null);verify(collection).overview(null);verify(collection).tasks(null);verify(collection).taskDetail("LXCOL-DEMO",null);verify(collection).runs(null); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopMasterDataMapperTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopMasterDataMapperTest.java new file mode 100644 index 000000000..09b4b8583 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/mapper/SelfopMasterDataMapperTest.java @@ -0,0 +1,60 @@ +package com.baiyue.lastleg.selfoperated.analytics.mapper; + +import com.baiyue.lastleg.selfoperated.analytics.model.SelfopPayloads; +import org.apache.ibatis.mapping.Environment; +import org.apache.ibatis.session.*; +import org.apache.ibatis.transaction.jdbc.JdbcTransactionFactory; +import org.h2.jdbcx.JdbcDataSource; +import org.junit.jupiter.api.*; +import java.sql.*; +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.nio.charset.StandardCharsets; +import java.util.UUID; +import static org.junit.jupiter.api.Assertions.*; + +/** Real SQL, source DDL and transaction rollback against an offline synthetic database. */ +class SelfopMasterDataMapperTest { + JdbcDataSource data;SqlSessionFactory factory;SqlSession session;SelfopMasterDataMapper mapper; + final LocalDateTime now=LocalDateTime.of(2026,10,14,8,0); + @BeforeEach void setup() throws Exception { + data=new JdbcDataSource();data.setURL("jdbc:h2:mem:"+UUID.randomUUID()+";MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1"); + try(var stream=getClass().getResourceAsStream("/db/selfop-master-data.sql")) { + String ddl=new String(stream.readAllBytes(),StandardCharsets.UTF_8).replace(" COLLATE utf8mb4_bin","").replaceAll(" ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='[^']*'",""); + for(String statement:ddl.split(";"))if(!statement.isBlank())sql(statement); + } + sql("CREATE TABLE ll_selfop_sync_task(id BIGINT PRIMARY KEY,source_account VARCHAR(64),task_type VARCHAR(32),task_role VARCHAR(32),execution_purpose VARCHAR(32))"); + sql("CREATE TABLE ll_selfop_sync_run(id BIGINT PRIMARY KEY,task_id BIGINT,source_account VARCHAR(64),execution_purpose VARCHAR(32),status VARCHAR(32),coverage_status VARCHAR(32),published_at_utc TIMESTAMP)"); + var config=new Configuration(new Environment("offline",new JdbcTransactionFactory(),data)); + config.setMapUnderscoreToCamelCase(true);config.setLocalCacheScope(LocalCacheScope.STATEMENT);config.addMapper(SelfopMasterDataMapper.class); + factory=new SqlSessionFactoryBuilder().build(config);session=factory.openSession(true);mapper=session.getMapper(SelfopMasterDataMapper.class); + } + @AfterEach void close(){session.close();} + void sql(String query)throws Exception{try(var c=data.getConnection();var s=c.createStatement()){s.execute(query);}} + SelfopPayloads.Product product(String owner){return new SelfopPayloads.Product(owner,"SKU","商品",new BigDecimal("5"),new BigDecimal("11"),new BigDecimal("5"),"cm",null,null,null,null);} + @Test void namesAndProductsAreIsolatedByAccountOwnerAndBatch()throws Exception{ + mapper.insertOwner(1,10,"A",new SelfopPayloads.Owner("C","客户","0"),now); + mapper.insertOwner(2,10,"B",new SelfopPayloads.Owner("C","另一账号","1"),now); + mapper.insertProduct(3,10,"A",product("C"),now);mapper.insertProduct(4,10,"A",product("D"),now); + assertEquals("客户",mapper.owners("A",10).get(0).customerName());assertEquals("0",mapper.owners("A",10).get(0).status()); + assertEquals(1,mapper.ownerCount(10,"A"));assertEquals(2,mapper.productCount(10,"A"));assertEquals(0,mapper.productCount(11,"A")); + assertThrows(RuntimeException.class,()->mapper.insertProduct(5,10,"A",product("C"),now)); + try(var c=data.getConnection();var s=c.createStatement();var rs=s.executeQuery("SELECT declared_length,declared_size_unit,measured_length,measured_size_unit FROM ll_selfop_product WHERE id=3")){ + assertTrue(rs.next());assertEquals(0,new BigDecimal("5").compareTo(rs.getBigDecimal(1)));assertEquals("cm",rs.getString(2));assertNull(rs.getBigDecimal(3));assertNull(rs.getString(4)); + } + } + @Test void onlyCompleteBusinessMasterBatchesPublishedBeforeAsOfAreVisible()throws Exception{ + sql("INSERT INTO ll_selfop_sync_task VALUES(1,'A','MASTER_DATA','ACCOUNT_DISCOVERY','BUSINESS'),(2,'A','BACKFILL','ACCOUNT_DISCOVERY','BUSINESS'),(3,'B','MASTER_DATA','ACCOUNT_DISCOVERY','BUSINESS')"); + sql("INSERT INTO ll_selfop_sync_run VALUES(10,1,'A','BUSINESS','PUBLISHED','COMPLETE','2026-10-14 07:00:00'),(11,1,'A','BUSINESS','PUBLISHED','COMPLETE','2026-10-14 09:00:00'),(12,1,'A','BUSINESS','FETCHING','COMPLETE','2026-10-14 07:59:00'),(13,1,'A','BUSINESS','PUBLISHED','PARTIAL','2026-10-14 07:59:00'),(14,2,'A','BUSINESS','PUBLISHED','COMPLETE','2026-10-14 07:59:00'),(15,3,'B','BUSINESS','PUBLISHED','COMPLETE','2026-10-14 07:59:00')"); + assertEquals(10L,mapper.publishedAt("A",now));assertEquals(11L,mapper.publishedAt("A",now.plusHours(2)));assertNull(mapper.publishedAt("A",now.minusHours(2)));assertEquals(15L,mapper.publishedAt("B",now)); + } + @Test void failedPageTransactionRollsBackEveryRow(){ + try(var transaction=factory.openSession(false)){ + var writer=transaction.getMapper(SelfopMasterDataMapper.class); + writer.insertOwner(1,10,"A",new SelfopPayloads.Owner("C","客户","0"),now); + assertThrows(RuntimeException.class,()->writer.insertOwner(2,10,"A",new SelfopPayloads.Owner("C","重复","0"),now)); + transaction.rollback(); + } + assertEquals(0,mapper.ownerCount(10,"A")); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryServiceTest.java index 257c65aa8..052ad51f4 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/AnalyticsQueryServiceTest.java @@ -16,6 +16,15 @@ import static org.mockito.Mockito.*; /** Offline contracts: SQL supplies bounded pages, never an in-memory warehouse history. */ class AnalyticsQueryServiceTest { + @Test void namesUseFrozenLocalDirectoryWithoutChangingSkuIdentityOrQuantities(){ + var directory=mock(SelfopOwnerDirectory.class);when(directory.names("demo",NOW)).thenReturn(Map.of("C","货主名称")); + var named=new AnalyticsQueryService(mapper,new ObjectMapper(),Clock.fixed(NOW,ZoneOffset.UTC),directory); + var before=service.inventoryCoverage(metrics(null),options()).records().get(0); + var after=named.inventoryCoverage(metrics(null),options()).records().get(0); + assertEquals("货主名称",after.customerName());assertEquals(before.customerCode(),after.customerCode());assertEquals(before.sku(),after.sku()); + assertEquals(before.physicalQuantity(),after.physicalQuantity());assertEquals(before.windowOutboundQuantity(),after.windowOutboundQuantity()); + assertEquals("货主名称",named.ownerDirectory(metrics(null)).get(0).customerName()); + } @Test void frozenHistoryStartExistsBeforeContinuousAnchorAndIncludesEarlierDisconnectedDays(){ when(mapper.publishedRun(11)).thenReturn(frozenRun(null,null, new com.baiyue.lastleg.selfoperated.analytics.model.SelfopSalesCoverage.Interval(time(D.minusDays(1)),time(D)))); @@ -398,7 +407,7 @@ class AnalyticsQueryServiceTest { return new Inventory("C","SKU","Product",unit,evidence,fresh?n("120"):null,stock,fresh?n("20"):null,fresh?n("30"):null, fresh?n("0"):null,utc(NOW.minusSeconds(3600)),out,out,10L,D.minusDays(1),evidence!=null,out,null,0,0,out==null||stock==null); } - private AnalyticsQueryService at(Instant now){return new AnalyticsQueryService(mapper,new ObjectMapper(),Clock.fixed(now,ZoneOffset.UTC));} + private AnalyticsQueryService at(Instant now){return new AnalyticsQueryService(mapper,new ObjectMapper(),Clock.fixed(now,ZoneOffset.UTC),org.mockito.Mockito.mock(SelfopOwnerDirectory.class));} private static AnalyticsQuery chart(String range,LocalDate from,LocalDate until,String owner,String sku){return new AnalyticsQuery("demo","W",owner,sku,range,from,until,"DAY",null,1,20,null,null,null,null,null);} private static AnalyticsQuery metrics(Integer days){return new AnalyticsQuery("demo","W",null,null,null,null,null,null,days,1,20,null,null,null,null,null);} private static AnalyticsQuery pinned(String range,long sales,long inventory,String sh,String ih,Instant asOf,int page){return new AnalyticsQuery("demo","W",null,null,range,null,null,range==null?null:"DAY",null,page,20,sales,inventory,sh,ih,asOf);} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardQueryServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardQueryServiceTest.java index d350178cd..1a4e1f466 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardQueryServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/DashboardQueryServiceTest.java @@ -13,7 +13,7 @@ import static org.mockito.Mockito.*; class DashboardQueryServiceTest { @Test void selectedWindowKeepsHotAvailableButDoesNotInventAverageWhenIncomplete()throws Exception{ - var contexts=mock(AnalyticsQueryService.class);var mapper=mock(DashboardReadMapper.class);var service=new DashboardQueryService(contexts,mapper); + var contexts=mock(AnalyticsQueryService.class);var mapper=mock(DashboardReadMapper.class);var service=new DashboardQueryService(contexts,mapper,org.mockito.Mockito.mock(SelfopOwnerDirectory.class)); LocalDate day=DashboardPolicyTest.DAY;Instant now=DashboardPolicyTest.NOW; var q=new AnalyticsQuery("demo","W",null,null,"TODAY",null,null,null,null,null,null,null,null,null,null,null); var context=DashboardPolicyTest.context(day.minusDays(30).atStartOfDay().toInstant(ZoneOffset.ofHours(8)),now,false); @@ -38,7 +38,7 @@ class DashboardQueryServiceTest { assertTrue(DashboardQueryService.class.getAnnotation(Transactional.class).readOnly()); } @Test void unreadyPublishedVersionIsRejectedWhileReadyEmptyWarehouseProvesZeroAndAbsentSalesStayUnknown(){ - var contexts=mock(AnalyticsQueryService.class);var mapper=mock(DashboardReadMapper.class);var service=new DashboardQueryService(contexts,mapper); + var contexts=mock(AnalyticsQueryService.class);var mapper=mock(DashboardReadMapper.class);var service=new DashboardQueryService(contexts,mapper,org.mockito.Mockito.mock(SelfopOwnerDirectory.class)); LocalDate day=DashboardPolicyTest.DAY;Instant now=DashboardPolicyTest.NOW; var q=new AnalyticsQuery("demo","W",null,null,"TODAY",null,null,null,null,null,null,null,null,null,null,null); var context=DashboardPolicyTest.context(day.minusDays(30).atStartOfDay().toInstant(ZoneOffset.ofHours(8)),now,false,day.minusDays(15),day); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContextTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContextTest.java index 4e8172201..be4349b77 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContextTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContextTest.java @@ -107,7 +107,7 @@ class InventoryTurnoverContextTest { when(analytics.applyBarriers("demo","W")).thenReturn(1L); when(mapper.report(any())).thenReturn(List.of(Map.of("section","TOTAL","payload","{\"total\":0}"), Map.of("section","OVERVIEW","payload","{\"sku_count\":0}"))); - var result=new InventoryTurnoverQueryService(contexts,mapper,json).query(request(null)); + var result=new InventoryTurnoverQueryService(contexts,mapper,json,org.mockito.Mockito.mock(SelfopOwnerDirectory.class)).query(request(null)); assertTrue(result.context().analyticsContext().rebuildRequired()); verify(mapper).report(argThat(p->p.days().stream().allMatch(d->d.coverage()==0))); verify(mapper).chartWindows(argThat(p->p.salesRun()==11L&&p.inventoryRun()==22L&&p.salesKnown() @@ -116,7 +116,7 @@ class InventoryTurnoverContextTest { &&p.days().subList(15,30).stream().allMatch(d->d.coverage()==1))); } @Test void firstPageSharesAggregationAndPaginationDoesNotRepeatOverviewQueries() { - var service=new InventoryTurnoverQueryService(contexts,mapper,json); + var service=new InventoryTurnoverQueryService(contexts,mapper,json,org.mockito.Mockito.mock(SelfopOwnerDirectory.class)); when(mapper.report(any())).thenReturn(List.of(Map.of("section","TOTAL","payload","{\"total\":0}"), Map.of("section","OVERVIEW","payload","{\"sku_count\":0,\"physical\":0,\"available\":0,\"no_outbound_count\":0,\"unassessable\":0}"))); var first=service.query(request(null));assertEquals(0,first.total());assertNotNull(first.summary());assertNotNull(first.charts()); @@ -127,7 +127,7 @@ class InventoryTurnoverContextTest { var q=new InventoryTurnoverRequest("demo","W",END,30,"SKU",null,null,"ALL",null,null,null,2,20,false,first.context().snapshotContext()); var next=service.query(q);assertNull(next.summary());assertNull(next.charts());verify(mapper).report(any());verify(mapper,never()).chartWindows(any()); } - private InventoryTurnoverContext at(Instant instant){return new InventoryTurnoverContext(analytics,mapper,json,Clock.fixed(instant,ZoneOffset.UTC));} + private InventoryTurnoverContext at(Instant instant){return new InventoryTurnoverContext(analytics,mapper,json,Clock.fixed(instant,ZoneOffset.UTC),mock(SelfopOwnerDirectory.class));} static InventoryTurnoverRequest request(String token){return new InventoryTurnoverRequest("demo","W",END,30,"SKU",null,null,"ALL",null,null,null,1,20,true,token);} private static LocalDateTime utc(Instant instant){return LocalDateTime.ofInstant(instant,ZoneOffset.UTC);} private static Scope scope(long id,String family,String hash,long run){return new Scope(id,"demo","W",family,hash,"Asia/Shanghai","Asia/Shanghai",93600,run,1L,1L,false,null,null,7);} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverQueryServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverQueryServiceTest.java new file mode 100644 index 000000000..7e0307d85 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverQueryServiceTest.java @@ -0,0 +1,21 @@ +package com.baiyue.lastleg.selfoperated.analytics.query; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import java.util.*; +import java.time.Instant; +import static org.mockito.Mockito.*; +import static org.junit.jupiter.api.Assertions.*; +class InventoryTurnoverQueryServiceTest { + @Test void totalsWithoutCustomerIdentityRemainReadableBeforeFirstDirectoryPublication(){ + var contexts=mock(InventoryTurnoverContext.class);var mapper=mock(InventoryTurnoverReadMapper.class);var directory=mock(SelfopOwnerDirectory.class); + var request=mock(InventoryTurnoverRequest.class);when(request.sourceAccount()).thenReturn("A"); + var context=mock(InventoryTurnoverResponses.Context.class);var analytics=mock(AnalyticsResponses.Context.class); + var snapshot=mock(AnalyticsQueryService.Snapshot.class);var asOf=Instant.parse("2026-10-14T08:00:00Z"); + when(context.queryAsOf()).thenReturn(asOf);when(context.analyticsContext()).thenReturn(analytics);when(context.dailySamples()).thenReturn(List.of()); + when(contexts.resolve(request)).thenReturn(new InventoryTurnoverContext.Frozen(context,snapshot)); + when(mapper.report(any())).thenReturn(List.of(Map.of("section","TOTAL","payload","{\"total\":0}"))); + var service=new InventoryTurnoverQueryService(contexts,mapper,new ObjectMapper(),directory); + when(directory.names("A",asOf)).thenReturn(Map.of());assertEquals(0,service.query(request).total()); + when(directory.names("A",asOf)).thenReturn(Map.of("C","货主名称"));assertEquals(0,service.query(request).total()); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/SelfopOwnerDirectoryTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/SelfopOwnerDirectoryTest.java new file mode 100644 index 000000000..e3a5dc4f4 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/SelfopOwnerDirectoryTest.java @@ -0,0 +1,17 @@ +package com.baiyue.lastleg.selfoperated.analytics.query; +import com.baiyue.lastleg.selfoperated.analytics.mapper.SelfopMasterDataMapper; +import com.baiyue.lastleg.selfoperated.analytics.model.SelfopPayloads; +import org.junit.jupiter.api.Test; +import java.time.*; +import java.util.*; +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; +class SelfopOwnerDirectoryTest { + @Test void onlyPublishedBatchAtFrozenTimeSuppliesNames(){ + var mapper=mock(SelfopMasterDataMapper.class);var time=Instant.parse("2026-10-14T08:00:00Z"); + when(mapper.publishedAt("A",LocalDateTime.ofInstant(time,ZoneOffset.UTC))).thenReturn(11L); + when(mapper.owners("A",11L)).thenReturn(List.of(new SelfopPayloads.Owner("C","客户","0"),new SelfopPayloads.Owner("D","",null))); + var names=new SelfopOwnerDirectory(mapper).names("A",time);assertEquals("客户",names.get("C"));assertTrue(names.containsKey("D"));assertNull(names.get("D")); + assertEquals(Map.of(),new SelfopOwnerDirectory(mapper).names("B",time));verify(mapper,never()).owners("B",11L); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java index dbb6257be..1468103ec 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java @@ -29,6 +29,25 @@ class SelfopEgressServiceTest { private void success(EgressContext c,Operation op){service.authorize(new AuthorizationRequest(c,op,PROVIDER));service.complete(new CompletionRequest(c,op,PROVIDER,1,null,200,200,false,null));clock.advance(5);} private SelfopApiBudget budget(){return SelfopSupport.read(account.getApiBudgetJson(),SelfopApiBudget.class);} private SelfopValidationBudget rootBudget(){return SelfopSupport.read(task.getValidationBudgetJson(),SelfopValidationBudget.class);} + @Test void masterDataOperationsUseBoundBusinessTaskAndExistingSingleFlightGate(){ + continuousBusinessRoot();task.setTaskType("MASTER_DATA");task.setQueryFromUtc(null);task.setQueryUntilUtc(null); + task.setPlannedAtUtc(now().plusHours(8).toLocalDate().atStartOfDay().minusHours(8)); + var definition=new TreeMap();definition.put("dataFamily","MASTER_DATA");definition.put("sourceAccount","synthetic"); + task.setQueryScopeJson(SelfopSupport.json(definition));task.setScopeHash(SelfopSupport.hash(definition)); + task.setRequestKey("continuous:master:"+task.getPlannedAtUtc().plusHours(8).toLocalDate()+":"+task.getScopeHash().substring(0,12)); + assertThat(SelfopCollectionPolicy.continuousBusinessRoot(task)).isTrue(); + assertThatThrownBy(()->reserve(Operation.OUTBOUND_QUERY,100,null,null)).hasMessage("EGRESS_OPERATION_TASK_MISMATCH"); + assertThatThrownBy(()->reserve(Operation.CUSTOMERS,1001,null,null)).hasMessage("EGRESS_OPERATION_TASK_MISMATCH"); + assertThatThrownBy(()->reserve(Operation.PRODUCTS,101,null,null)).hasMessage("EGRESS_OPERATION_TASK_MISMATCH"); + var c=reserve(Operation.CUSTOMERS,1000,null,null); + assertThatThrownBy(()->reserve(Operation.PRODUCTS,100,null,null)).isInstanceOf(EgressRejectedException.class); + success(c,Operation.CUSTOMERS);success(reserve(Operation.PRODUCTS,100,null,null),Operation.PRODUCTS); + assertThat(budget().confirmedHttp).isEqualTo(2); + } + @Test void ordinaryTasksCannotUseMasterDataOperations(){ + business();assertThatThrownBy(()->reserve(Operation.CUSTOMERS,10,null,null)).hasMessage("EGRESS_OPERATION_TASK_MISMATCH"); + assertThatThrownBy(()->reserve(Operation.PRODUCTS,10,null,null)).hasMessage("EGRESS_OPERATION_TASK_MISMATCH"); + } @Test void offUnknownModeUnknownEnvironmentBusinessAndUnverifiedHandoffFailBeforeAnyLock(){ cfg.setCollectionMode("OFF");assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EGRESS_MODE_REJECTED"); cfg.setCollectionMode("BUSINESS");cfg.setEnvironment("unknown");assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("BUSINESS_ENVIRONMENT_REJECTED"); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopContinuousDueSqlTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopContinuousDueSqlTest.java index ab082a463..58e41ae3c 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopContinuousDueSqlTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopContinuousDueSqlTest.java @@ -44,6 +44,12 @@ class SelfopContinuousDueSqlTest { try(var rows=p.executeQuery()){var result=new ArrayList();while(rows.next())result.add(rows.getLong("id"));return result;} } } + @Test void masterCatalogIsDispatchableAfterInventoryAndHistoricalSales() throws Exception { + inventory(2,"READY",0,MIDNIGHT,null); + execute("INSERT INTO ll_selfop_sync_task VALUES(1,1,'SYNTH-ACCOUNT','BUSINESS',NULL,'READY','continuous:master:day','SELFOP_CONTINUOUS_PLANNER','ACCOUNT_DISCOVERY','MASTER_DATA',NULL,?,NULL,0)",MIDNIGHT.minusDays(1)); + execute("INSERT INTO ll_selfop_sync_task VALUES(3,3,'SYNTH-ACCOUNT','BUSINESS',NULL,'READY','continuous:outbound:day','SELFOP_CONTINUOUS_PLANNER','ACCOUNT_DISCOVERY','BACKFILL',NULL,NULL,?,0)",MIDNIGHT.minusDays(10)); + assertEquals(List.of(2L,3L,1L),due(MIDNIGHT)); + } @Test void retryAfterBeijingMidnightWaitsUntilBackoffThenRemainsDispatchable() throws Exception { // Failure at 23:59:55 + the first 10-second backoff lands on the next Beijing day. inventory(1,"RETRY_WAIT",1,MIDNIGHT.minusHours(16),MIDNIGHT.plusSeconds(5)); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionServiceTest.java index 48f6ddcaa..9bffadefe 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopExecutionServiceTest.java @@ -20,7 +20,7 @@ class SelfopExecutionServiceTest { when(leases.acquireScopes(any(),anyString())).thenReturn(java.util.Map.of(3L,lease));when(taskService.begin(1L,lease)).thenReturn(run); var candidates=mock(OutboundCandidateBuilder.class);var publisher=mock(SelfopSalesPublisher.class);var runs=mock(SelfopSyncRunMapper.class); when(candidates.reusable(anyLong(),anyLong(),anyString(),anyLong())).thenReturn(null); - var service=new SelfopExecutionService(config,taskService,leases,mock(SelfopDiscoveryService.class),mock(SelfopInventoryCollector.class),mock(InventorySnapshotService.class),mock(OutboundFactService.class),candidates,publisher,scopes,tasks,mock(SelfopSyncStageMapper.class),mock(SelfopOutboundOrderMapper.class),mock(SelfopReferenceRecovery.class),mock(SelfopValidationRunner.class),runs); + var service=new SelfopExecutionService(config,taskService,leases,mock(SelfopDiscoveryService.class),mock(SelfopInventoryCollector.class),mock(InventorySnapshotService.class),mock(OutboundFactService.class),candidates,publisher,scopes,tasks,mock(SelfopSyncStageMapper.class),mock(SelfopOutboundOrderMapper.class),mock(SelfopReferenceRecovery.class),mock(SelfopValidationRunner.class),runs,org.mockito.Mockito.mock(SelfopMasterDataCollector.class)); service.execute(1L); var order=inOrder(candidates,publisher);order.verify(candidates).build(eq(3L),eq(2L),anyString(),eq(8L));order.verify(publisher).publish(eq(3L),eq(2L),anyString(),eq(8L)); verify(taskService,never()).waitForBudget(anyLong(),anyLong(),any(),any());verifyNoInteractions(runs); @@ -48,7 +48,7 @@ class SelfopExecutionServiceTest { when(taskService.begin(1L,lease)).thenReturn(run);doThrow(failure).when(validation).run(task,run,lease); var service=new SelfopExecutionService(config,taskService,leases,discovery,stock,snapshots,facts,candidates,publisher, mock(SelfopSyncScopeMapper.class),tasks,mock(SelfopSyncStageMapper.class),mock(SelfopOutboundOrderMapper.class), - recovery,validation,mock(SelfopSyncRunMapper.class)); + recovery,validation,mock(SelfopSyncRunMapper.class),org.mockito.Mockito.mock(SelfopMasterDataCollector.class)); service.execute(1L); verify(validation).fail(task,run,lease,code);verify(leases).release(lease); verify(taskService,never()).fail(anyLong(),anyLong(),anyString(),anyBoolean(),any()); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollectorTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollectorTest.java new file mode 100644 index 000000000..9d9fc56c6 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopMasterDataCollectorTest.java @@ -0,0 +1,76 @@ +package com.baiyue.lastleg.selfoperated.analytics.sync; + +import com.baiyue.lastleg.selfoperated.analytics.client.LingxingAnalyticsClient; +import com.baiyue.lastleg.selfoperated.analytics.mapper.*; +import com.baiyue.lastleg.selfoperated.analytics.model.*; +import org.junit.jupiter.api.*; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.dao.DuplicateKeyException; +import java.util.*; +import java.util.function.Consumer; +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; +import static org.mockito.ArgumentMatchers.*; + +class SelfopMasterDataCollectorTest { + final LingxingAnalyticsClient client=mock(LingxingAnalyticsClient.class); + final SelfopMasterDataMapper data=mock(SelfopMasterDataMapper.class); + final SelfopSyncTaskMapper tasks=mock(SelfopSyncTaskMapper.class); + final SelfopSyncRunMapper runs=mock(SelfopSyncRunMapper.class); + final SelfopTaskService service=mock(SelfopTaskService.class); + final SelfopLeaseService leases=mock(SelfopLeaseService.class); + final SelfopAnalyticsProperties config=new SelfopAnalyticsProperties(); + final TransactionTemplate tx=mock(TransactionTemplate.class); + final SelfopSyncTask task=new SelfopSyncTask();final SelfopSyncRun run=new SelfopSyncRun(); + final SelfopLeaseService.Lease lease=new SelfopLeaseService.Lease(1L,"owner",2L,true); + SelfopMasterDataCollector collector; + @BeforeEach void setup(){ + config.setEnvironment("production");task.setId(10L);task.setTaskType("MASTER_DATA");task.setExecutionPurpose("BUSINESS");task.setSourceAccount("A");task.setTaskNo("LXCOL-DEMO"); + run.setId(20L);run.setTaskId(10L);run.setExecutionPurpose("BUSINESS");run.setStatus("FETCHING");run.setAttemptNo(1); + when(service.checkpoint(task,run,lease)).thenReturn(new SelfopTaskService.CollectionCheckpoint(task,run)); + doAnswer(i->{Consumer c=i.getArgument(0);c.accept(mock(TransactionStatus.class));return null;}).when(tx).executeWithoutResult(any()); + when(client.customers(any(),eq(1),eq(1000))).thenReturn(new LingxingAnalyticsClient.Page<>(List.of(new SelfopPayloads.Owner("C","客户","0")),1,1000,1L,1)); + when(data.ownerCount(20L,"A")).thenReturn(1L); + collector=new SelfopMasterDataCollector(client,data,tasks,runs,service,leases,config,tx); + } + SelfopPayloads.Product product(String code,String sku){return new SelfopPayloads.Product(code,sku,"商品",null,null,null,"cm",null,null,null,null);} + @Test void publishesBothDirectoriesOnlyAfterBothCountsMatch(){ + when(client.products(any(),eq(1),eq(100))).thenReturn(new LingxingAnalyticsClient.Page<>(List.of(product("C","SKU"),product("D","SKU")),1,100,2L,1)); + when(data.productCount(20L,"A")).thenReturn(2L); + collector.collect(task,run,lease); + assertEquals("PUBLISHED",run.getStatus());assertEquals("SUCCEEDED",task.getStatus());assertEquals(3L,run.getUniqueObjectCount()); + verify(data,times(2)).insertProduct(anyLong(),eq(20L),eq("A"),any(),any());verify(service,never()).yield(anyLong(),anyLong(),any()); + } + @Test void countMismatchCannotPublish(){ + when(client.products(any(),eq(1),eq(100))).thenReturn(new LingxingAnalyticsClient.Page<>(List.of(product("C","SKU")),1,100,1L,1)); + assertEquals("SELFOP_MASTER_COUNT_MISMATCH",assertThrows(IllegalStateException.class,()->collector.collect(task,run,lease)).getMessage()); + assertNotEquals("PUBLISHED",run.getStatus()); + } + @Test void duplicateFailsInsteadOfDeduplicatingPartialCatalog(){ + doThrow(new DuplicateKeyException("duplicate")).when(data).insertOwner(anyLong(),anyLong(),anyString(),any(),any()); + assertEquals("SELFOP_MASTER_DUPLICATE",assertThrows(IllegalStateException.class,()->collector.collect(task,run,lease)).getMessage()); + verify(client,never()).products(any(),anyInt(),anyInt()); + } + @Test void lostFenceCannotPersistLateResponse(){ + when(service.checkpoint(task,run,lease)).thenThrow(new IllegalStateException("FENCE_LOST")); + assertThrows(IllegalStateException.class,()->collector.collect(task,run,lease));verifyNoInteractions(data); + } + @Test void yieldsAndResumesSameBatchWithoutRepeatingCustomerPage(){ + for(int page=1;page<=6;page++){ + var rows=new ArrayList();for(int n=0;n<100;n++)rows.add(product("C","SKU-"+page+"-"+n)); + when(client.products(any(),eq(page),eq(100))).thenReturn(new LingxingAnalyticsClient.Page<>(rows,page,100,600L,6)); + } + when(data.productCount(20L,"A")).thenReturn(600L); + collector.collect(task,run,lease);verify(service).yield(10L,20L,lease);assertNotEquals("PUBLISHED",run.getStatus()); + collector.collect(task,run,lease);assertEquals("PUBLISHED",run.getStatus());verify(client,times(1)).customers(any(),eq(1),eq(1000)); + verify(data,times(600)).insertProduct(anyLong(),eq(20L),eq("A"),any(),any()); + } + @Test void changedTotalRejectsBeforePersistence(){ + var manifest=new SelfopMasterDataCollector.Manifest(); + var rows=Collections.nCopies(100,product("C","SKU")); + SelfopMasterDataCollector.validatePage(manifest,new LingxingAnalyticsClient.Page<>(rows,1,100,201L,3),100); + manifest.setNextPage(2); + assertEquals("SELFOP_PAGE_TOTAL_CHANGED",assertThrows(IllegalStateException.class,()->SelfopMasterDataCollector.validatePage(manifest,new LingxingAnalyticsClient.Page<>(rows,2,100,202L,3),100)).getMessage()); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java index 91b1f9089..0ef8cf949 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlannerTest.java @@ -16,6 +16,17 @@ import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; class SelfopSchedulePlannerTest { + @Test void masterCatalogIsOneAccountBatchAndAnUnfinishedBatchSpansDays(){ + var state=continuousState(); + planner.planContinuous(LocalDateTime.of(2026,10,13,4,0)); + var master=state.stream().filter(t->"MASTER_DATA".equals(t.getTaskType())).findFirst().orElseThrow(); + assertThat(SelfopCollectionPolicy.continuousBusinessRoot(master)).isTrue();assertThat(master.getWarehouseCode()).isNull(); + planner.planContinuous(LocalDateTime.of(2026,10,14,4,0)); + assertThat(state.stream().filter(t->"MASTER_DATA".equals(t.getTaskType()))).hasSize(1); + master.setStatus("SUCCEEDED");planner.planContinuous(LocalDateTime.of(2026,10,14,4,1)); + planner.planContinuous(LocalDateTime.of(2026,10,14,4,2)); + assertThat(state.stream().filter(t->"MASTER_DATA".equals(t.getTaskType()))).hasSize(2); + } @Test void continuousInventoryWaitsForTodaysSlotAndNeverBackfillsPreviousDays(){ var state=continuousState(); var manual7=dayRoot("RUNNING",7,"manual:7");var manual6=dayRoot("READY",6,"manual:6"); @@ -50,15 +61,15 @@ class SelfopSchedulePlannerTest { @Test void continuousPoolPersistsCompleteDaysNewestFirstAndNeverChangesCoverageAnchor(){ var state=continuousState(); planner.planContinuous(LocalDateTime.of(2026,10,13,4,0)); - assertThat(state).hasSize(62); + assertThat(state).hasSize(63); assertThat(state.get(0).getQueryFromUtc()).isEqualTo(LocalDateTime.of(2026,10,11,16,0)); assertThat(state.get(60).getQueryFromUtc()).isEqualTo(LocalDateTime.of(2026,8,12,16,0)); assertThat(state).allMatch(SelfopCollectionPolicy::continuousBusinessRoot); verify(scopes,never()).updateById(any(SelfopSyncScope.class));verify(taskService,never()).initialize(); - planner.planContinuous(LocalDateTime.of(2026,10,13,4,1));assertThat(state).hasSize(62); - planner.planContinuous(LocalDateTime.of(2026,10,13,16,4));assertThat(state).hasSize(62); - planner.planContinuous(LocalDateTime.of(2026,10,13,16,5));assertThat(state).hasSize(63); - assertThat(state.get(62).getQueryFromUtc()).isEqualTo(LocalDateTime.of(2026,10,12,16,0)); + planner.planContinuous(LocalDateTime.of(2026,10,13,4,1));assertThat(state).hasSize(63); + planner.planContinuous(LocalDateTime.of(2026,10,13,16,4));assertThat(state).hasSize(63); + planner.planContinuous(LocalDateTime.of(2026,10,13,16,5));assertThat(state).hasSize(64); + assertThat(state.get(63).getQueryFromUtc()).isEqualTo(LocalDateTime.of(2026,10,12,16,0)); assertThat(state).anyMatch(t->LocalDateTime.of(2026,8,12,16,0).equals(t.getQueryFromUtc())); } @Test void deadOldManualRootDoesNotMaskSuccessfulOrActiveDayAndContinuousFailureIsNotRecreated(){ @@ -67,7 +78,7 @@ class SelfopSchedulePlannerTest { state.add(dayRoot("RUNNING",11,"manual:running")); state.add(dayRoot("DEAD",10,"manual:failed-only")); planner.planContinuous(LocalDateTime.of(2026,10,13,4,0)); - assertThat(state.stream().filter(SelfopCollectionPolicy::continuousBusinessRoot)).hasSize(60); + assertThat(state.stream().filter(SelfopCollectionPolicy::continuousBusinessRoot)).hasSize(61); assertThat(state.stream().filter(SelfopCollectionPolicy::continuousBusinessRoot)).noneMatch(t->LocalDateTime.of(2026,10,11,16,0).equals(t.getQueryFromUtc())); var root=state.stream().filter(SelfopCollectionPolicy::continuousBusinessRoot).findFirst().orElseThrow();root.setStatus("DEAD"); int count=state.size();planner.planContinuous(LocalDateTime.of(2026,10,13,4,1));assertThat(state).hasSize(count); @@ -93,7 +104,12 @@ class SelfopSchedulePlannerTest { return query.getParamNameValuePairs().containsValue("INVENTORY")?List.of(stock):List.of(scope);}); var account=new SelfopOutboundDiscoveryScope();account.setId(30L);when(taskService.initializeAccount()).thenReturn(account); var state=new ArrayList();when(tasks.selectList(any())).thenAnswer(call->new ArrayList<>(state)); - when(taskService.create(any())).thenAnswer(call->{SelfopSyncTask task=call.getArgument(0);state.add(task);return task;});return state; + when(tasks.selectCount(any())).thenAnswer(call->{ + com.baomidou.mybatisplus.core.conditions.query.QueryWrapper query=call.getArgument(0);query.getSqlSegment(); + var values=query.getParamNameValuePairs().values(); + return state.stream().filter(t->"MASTER_DATA".equals(t.getTaskType())).filter(t->values.contains("MASTER_DATA")?List.of("READY","RUNNING","RETRY_WAIT").contains(t.getStatus()):values.contains(t.getRequestKey())).count(); + }); + when(taskService.create(any())).thenAnswer(call->{SelfopSyncTask task=call.getArgument(0);task.setStatus("READY");state.add(task);return task;});return state; } private SelfopSyncTask dayRoot(String status,int day,String key){ var root=controlled(100L+day,"ACCOUNT_DISCOVERY","BACKFILL",key);root.setDiscoveryScopeId(30L);root.setScopeHash("a".repeat(64));root.setStatus(status);