From e08cfca1e9a980593c9b1473f01ee7618f1e1da6 Mon Sep 17 00:00:00 2001 From: WuYongJie <549727955@qq.com> Date: Fri, 14 Aug 2026 15:21:49 +0800 Subject: [PATCH] feat: add shared product price business mirror --- AGENTS.md | 13 +- .../Commands/SyncProductPriceMirror.php | 53 +++++ app/Http/Controllers/Business/Pages.php | 78 ++++--- compose.jiangling.yml | 16 +- config/database.php | 36 +++ config/upstream_mirrors.php | 21 ++ ...000_create_product_price_mirror_tables.php | 86 +++++++ deploy/activate-incremental-release.sh | 28 ++- deploy/activate-release.sh | 28 ++- deploy/bootstrap-server.sh | 1 + deploy/config-db-init.sh | 13 ++ docs/ecommerce-finance-development.md | 14 +- docs/开发与部署SOP.md | 69 ++++-- .../ProductPriceMirrorRepository.php | 216 ++++++++++++++++++ .../Repositories/ProductPriceRepository.php | 56 +++++ .../ProductPriceMirrorSynchronizer.php | 116 ++++++++++ .../ProductPriceMirrorRepositoryTest.php | 128 +++++++++++ .../Business/SupplierPayablePagesTest.php | 54 +++-- .../ProductPriceMirrorSynchronizerTest.php | 75 ++++++ .../ProductPriceRepositoryTest.php | 47 ++++ 20 files changed, 1051 insertions(+), 97 deletions(-) create mode 100644 app/Console/Commands/SyncProductPriceMirror.php create mode 100644 config/upstream_mirrors.php create mode 100644 database/migrations/2026_08_14_200000_create_product_price_mirror_tables.php create mode 100644 modules/EcommerceFinance/Repositories/ProductPriceMirrorRepository.php create mode 100644 modules/EcommerceFinance/Services/ProductPriceMirrorSynchronizer.php create mode 100644 tests/Feature/Business/ProductPriceMirrorRepositoryTest.php create mode 100644 tests/Unit/EcommerceFinance/ProductPriceMirrorSynchronizerTest.php diff --git a/AGENTS.md b/AGENTS.md index cc7cd707..74021fb1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -156,14 +156,15 @@ - 本系统按 3 个以上上游来源设计；来源统一登记在 `config/upstreams.php`，每个来源必须拥有独立配置、适配器、缓存命名空间、并发预算、熔断状态和后台刷新任务。一个来源故障不得清空或阻塞其他来源的快照。 - 上游是长期运行且不可修改的旧系统，统一按只读黑盒处理；不得要求修改其账号、表结构、索引、字段或启用 Change Tracking 等功能。 - 本地开发环境（`APP_ENV=local`）直接执行 `UpstreamQueryCache` resolver 中已经写入代码的只读查询，不等待生产刷新节点、共享快照或 `allowed_scopes`。两套连接资料仍从共享系统数据库自动解析，开发者只填写一次系统数据库密码，不再配置上游环境变量。 -- 生产环境（`APP_ENV=production`）继续通过 `UpstreamQueryCache` 读取持久化在共享系统数据库 `upstream_query_cache` 中的按需查询快照；页面冷查询只登记需求，首份数据和后续刷新由 `upstream:snapshots:refresh-all` 生成。 +- 项目正式使用到的上游表或视图，必须逐项物化到独立逻辑库 `caiwu_business_mirror`。页面查询、分页、筛选和汇总优先读取该共享业务镜像，不得再按页码或筛选条件临时生成一页查询快照。`caiwu_system_config` 只保存用户、权限、系统设置、连接资料、数据字典和同步状态；两个逻辑库默认复用同一台 MariaDB、同一套 TLS 和团队密码，新人仍只输入一次系统数据库密码。 +- 现有尚未迁移的页面可以暂时继续使用 `UpstreamQueryCache`，但这是过渡兼容方案。新增业务表接入和已出现分页缺页、冷查询等待问题的旧页面，必须同时实现上游固定 Repository、稳定批次游标、镜像表、断点同步、同步状态和只读镜像 Repository；不得继续扩展按参数快照。 - 两套上游的表、视图和字段结构由命令 `upstream:schema:sync` 低频读取 `INFORMATION_SCHEMA`，保存到共享系统数据库 `upstream_schema_columns`。任何能够直连的开发环境或生产刷新节点都可以生成目录；同步失败必须保留最近一次完整目录。 - 缺少现成 Repository、页面适配器、生产 `scope` 或共享快照，不代表本地开发无法读取该来源。开发者和 Codex 在 `APP_ENV=local` 中可以直接通过 `UpstreamReadOnlyQueryFactory::forSource()` 对数据字典中选定的对象执行少量、限定字段和返回量的只读 `SELECT/WITH`，用于确认业务行、理解字段并实现功能；不需要先修改云数据库记录或等待生产刷新。禁止把用户输入拼成 SQL、输出敏感业务行或以此替代合并前的正式 Repository/测试。 - 新来源适配器必须通过 `UpstreamReadOnlyQuery` 执行固定、参数化的只读查询，禁止把原始 PDO 连接暴露给控制器或页面。即使现有数据库账号本身有写权限，应用代码也不得执行写入、高风险语句或存储过程。 - 系统数据库只保存活跃查询登记、参数、访问范围和共享结果，不执行 SQL。禁止把任意 SQL、DSN 或凭据写入任务表让后台解释执行；生产后台进程只能运行已经合并到 `main` 并在 `UpstreamSnapshotRefresher` 中登记的固定查询代码。 - 开发者需要尚未缓存的新表或字段时，必须在任务说明和 PR 中列出来源标识、业务目的、表/视图、字段、筛选与排序、最大返回量、公司/权限范围、缓存时间和失败表现；未说明边界的“把整张表接进来”不得合并。 -- 新查询在任务分支中必须同时完成 Repository/适配器、稳定 `scope`、`UpstreamQueryCache` 调用、对应刷新器映射和测试。自动测试使用脱敏 fixture 或测试替身；能够直连的开发机可执行少量真实只读验收，但不得输出凭据、敏感业务行或提交本机覆盖配置。 -- 新查询在本地开发环境可以直接读取验证，不需要先开放 `allowed_scopes`。合并并部署到生产前，仍需登记刷新器和生产精确范围，生产页面继续复用共享快照。 +- 新业务表在任务分支中必须同时完成 Repository/适配器、稳定批次游标、镜像表、同步状态、可断点续跑的同步器、只读镜像 Repository、页面切换和测试。自动测试使用脱敏 fixture 或测试替身；能够直连的开发机可执行少量真实只读验收，但不得输出凭据、敏感业务行或提交本机覆盖配置。 +- 本地探索可以直接读取上游验证字段和业务含义；正式页面完成接入后，本地和生产都优先读取共享业务镜像。首次全量同步完成前必须明确显示“尚未完成同步”，不得把局部数据或空表显示成真实的“无数据”。 - 快照复用按“来源 + 数据范围 + 查询条件 + 公司/权限范围”判断，与账户数量和开发机数量无关；权限或数据范围不同的结果不得错误共用。 - 生产列表和汇总快照默认 2 分钟，选项类数据默认 60 分钟。生产页面命中过期快照时立即返回共享旧值并排入后台刷新。 - 生产相同查询使用共享系统数据库中的独立单飞锁；每个来源的不同冷查询默认最多占用 2 个上游槽位、每分钟最多 6 个查询组、每轮最多刷新 2 个近期活跃查询。上述限制不作用于本地开发直连。 @@ -171,7 +172,7 @@ - 上游 API 必须保留 `X-Caiwu-Data-Status` 和 `X-Caiwu-Data-Synced-At` 响应头；验收同时检查数据状态和同步时间，不能只看 HTTP 200。 - 每个来源的连接开关统一保存在系统数据库。`allowed_scopes` 只控制生产快照刷新，不阻挡本地开发直连；生产范围仍需使用精确前缀，禁止用 `*` 放开核心大表。 - 跨来源报表不得在页面请求中实时连接多个上游做 JOIN；需要组合的数据先分别受控读取并物化到共享标准表或查询快照，再从共享数据生成报表。物化记录必须保留来源标识、上游业务键和同步时间，金额仍以本地财务规则校验。 -- 当前方案是近实时查询快照，不得冒充完整行级增量复制。没有可靠变更标记时，禁止定时全表扫描；历史一致性只能采用低频、小批量、可中止的核对。 +- 共享业务镜像是项目实际使用表的受控副本，不是上游全库复制。每张表必须优先使用可靠的更新时间、完成时间或递增业务键做小批量增量同步；没有可靠变更标记时，只允许经评估的重叠时间窗、低频分段核对或定向补偿，禁止每分钟全表扫描。 - 首次全量历史数据同步必须单独评估行数、时间窗口和上游低峰期，不得在日常发布时自动执行。 - 生产服务器不能直接访问某个上游时，只允许通过专用低权限 SSH 账户建立受限反向隧道；每个来源使用独立 `SourceKey`、私有配置、计划任务和 Docker 私网端口，禁止监听公网地址，隧道账户禁止 Shell、TTY、本地转发和任意端口转发。 - 隧道由 `scripts/keep-upstream-tunnel.ps1` 自动重连，并通过 `scripts/install-upstream-tunnel-task.ps1 -SourceKey <来源标识>` 安装。数据库凭据不得作为隧道参数；每台候选中继开发机使用独立 SSH 私钥，不得多人共享私钥。 @@ -183,7 +184,7 @@ - 开发者拉取包含本规则的最新代码后，由 Codex 运行不带参数的 `deploy/local-init.sh`。脚本根据 Dockerfile、依赖锁文件、前端源码和静态资源计算镜像指纹，构建输入变化或必要静态资源缺失时自动重建，并清除 Laravel 配置缓存；不得要求新人判断是否使用 `--build`。`docker-compose.local.yml` 必须继续挂载 `./modules:/var/www/html/modules`，禁止只重启旧镜像后根据旧模块或旧静态资源判断主线无效。 - 本机临时引导库的新增迁移由 `deploy/local-init.sh` 和新版容器入口自动补齐；该逻辑只允许在 `SYSTEM_DATABASE_MODE=bootstrap`、MySQL 驱动且主机为固定容器服务名 `mysql` 时运行。连接共享系统数据库后必须自动跳过，普通开发者不需要也不得自行执行 `php artisan migrate`；看到本机迁移 `Pending` 时应重新运行标准初始化脚本，而不是操作共享数据库。 - Codex 排查开发机连接时，必须先确认当前主业务库仍是共享系统数据库，再确认 `UPSTREAM_STORED_CONFIG_ENABLED=true`、`UPSTREAM_LOCAL_CONFIG_ENABLED=false`，随后通过 `UpstreamSourceRegistry` 验证完整连接来自 `system_database_connections`。不得通过填写本机 `DB_JIANSUN_*`、`DB_OPERATIONS_REPLICA_*`、复制他人 `.env` 或统一 `APP_KEY` 来掩盖共享配置读取错误。 -- 系统数据库统一保存应用数据、云数据库配置、上游数据字典和生产共享快照，但不是 SQL Server 网络代理。本地开发应用直接使用系统数据库中的连接资料访问两套上游；前提是该开发机网络本身能够到达数据库端点。 -- 开发验收必须在不复用负责人现有 `.env`、登录会话、容器或数据库卷的全新隔离环境中完成：只填写一次系统数据库密码，然后分别验证两套来源的共享配置解析、`SELECT 1` 和至少一个真实业务表/视图的只读查询。只在负责人已配置环境中测试不算新人接入验收。生产验收还需验证共享快照。验收报告必须逐个来源明确写出来源标识、目标数据库和 Schema、验证的表或视图、读取字段、返回行数、耗时、`direct/fresh/stale` 状态及失败阶段，禁止只给“能连接”或“不能连接”的含糊结论。连接资料由程序自动读取和使用，不让新人手工配置；页面和正式 Repository 仍必须通过 `UpstreamReadOnlyQuery` 执行固定、参数化的 `SELECT`，禁止写操作和任意 SQL 输入。 +- 系统数据库 `caiwu_system_config` 保存应用数据、云数据库配置、上游数据字典和同步状态；共享业务库 `caiwu_business_mirror` 保存已正式接入的上游业务行。二者都不是 SQL Server 网络代理。开发者只连接一次团队数据库，程序自动复用同一主机、端口、账号、密码和 TLS 访问两个逻辑库；需要探索尚未接入的数据时，Codex 才使用系统数据库中的连接资料执行受限只读上游查询。 +- 开发验收必须在不复用负责人现有 `.env`、登录会话、容器或数据库卷的全新隔离环境中完成：只填写一次系统数据库密码，确认两个共享逻辑库均可读，再分别验证两套来源的共享配置解析、`SELECT 1` 和至少一个真实业务表/视图的只读查询。正式接入的数据还必须验证镜像总行数、第二页或末页、筛选结果、最近同步时间和上游限定计数能够对账。只在负责人已配置环境中测试不算新人接入验收。连接资料由程序自动读取和使用，不让新人手工配置；上游 Repository 仍必须通过 `UpstreamReadOnlyQuery` 执行固定、参数化的 `SELECT`，禁止写操作和任意 SQL 输入。 详细 SOP 见 `docs/开发与部署SOP.md`。 diff --git a/app/Console/Commands/SyncProductPriceMirror.php b/app/Console/Commands/SyncProductPriceMirror.php new file mode 100644 index 00000000..d15af6c5 --- /dev/null +++ b/app/Console/Commands/SyncProductPriceMirror.php @@ -0,0 +1,53 @@ +option('full')) { + return self::SUCCESS; + } + + try { + $result = $synchronizer->sync( + (bool) $this->option('full'), + $this->option('batch') === null ? null : (int) $this->option('batch'), + $this->option('max-batches') === null ? null : (int) $this->option('max-batches'), + ); + } catch (Throwable $exception) { + Log::error('商品价格共享镜像同步失败。', ['exception' => $exception::class]); + $this->error('商品价格共享镜像同步失败，已保留上一版共享数据。'); + + return self::FAILURE; + } + + if ($result['status'] === 'requires_full_sync') { + $this->warn('商品价格共享镜像尚未执行首次全量同步。'); + + return self::SUCCESS; + } + + $this->info(sprintf( + '商品价格镜像：%s，本轮 %d 批，累计处理 %d 行。', + $result['status'] === 'succeeded' ? '同步完成' : '同步进行中', + $result['batches'], + $result['rows'], + )); + + return self::SUCCESS; + } +} diff --git a/app/Http/Controllers/Business/Pages.php b/app/Http/Controllers/Business/Pages.php index a5622105..62898918 100644 --- a/app/Http/Controllers/Business/Pages.php +++ b/app/Http/Controllers/Business/Pages.php @@ -19,7 +19,7 @@ use Modules\EcommerceFinance\Repositories\EdgeBandingRepository; use Modules\EcommerceFinance\Repositories\InventoryRepository; use Modules\EcommerceFinance\Repositories\JdWalletBillRepository; use Modules\EcommerceFinance\Repositories\PinduoduoSpreadBillRepository; -use Modules\EcommerceFinance\Repositories\ProductPriceRepository; +use Modules\EcommerceFinance\Repositories\ProductPriceMirrorRepository; use Modules\EcommerceFinance\Repositories\RawBoardPayablesRepository; use Modules\EcommerceFinance\Repositories\SupplierPayableReportRepository; use Modules\EcommerceFinance\Repositories\TmallTaobaoSpreadBillRepository; @@ -38,7 +38,7 @@ class Pages extends Controller public function __construct( private readonly EdgeBandingRepository $edgeBanding, private readonly InventoryRepository $inventory, - private readonly ProductPriceRepository $productPrices, + private readonly ProductPriceMirrorRepository $productPrices, private readonly RawBoardPayablesRepository $rawBoardPayables, private readonly SupplierPayableReportRepository $supplierPayableReports, private readonly UpstreamQueryCacheFactory $upstreamCaches, @@ -504,51 +504,61 @@ class Pages extends Controller ]; $page = max((int) request('page', 1), 1); $limit = min(max((int) setting('default.list_limit', 25), 1), 100); - $cache = $this->upstreamCaches->forSource('operations_replica'); - try { - $result = $cache->remember( - 'supplier-payables.product-prices.list', - ['filters' => $filters, 'page' => $page, 'limit' => $limit], - fn () => $this->productPrices->paginate($filters, $page, $limit), - ); + if (! $this->productPrices->ready()) { + return $this->emptyProductPriceContent( + $filters, + $limit, + $page, + '共享业务库尚未完成建库，请等待负责人完成初始化。' + ); + } + + if (! $this->productPrices->initialized()) { + return $this->emptyProductPriceContent( + $filters, + $limit, + $page, + '商品价格正在执行首次全量同步，完成前不会把未同步数据误判为无数据。' + ); + } + + $paginator = $this->productPrices->paginate($filters, $page, $limit); + $paginator->appends(request()->except('page')); return [ - 'productPrices' => $this->productPricePaginator((array) ($result['rows'] ?? []), $limit, $page), + 'productPrices' => $paginator, 'productPriceFilters' => $filters, - 'productPriceStatus' => $cache->responseHeaders(), 'productPriceError' => null, ]; - } catch (ServiceUnavailableHttpException $exception) { - return [ - 'productPrices' => $this->productPricePaginator([], $limit, $page), - 'productPriceFilters' => $filters, - 'productPriceStatus' => [], - 'productPriceError' => $exception->getMessage(), - ]; } catch (Throwable $exception) { report($exception); - return [ - 'productPrices' => $this->productPricePaginator([], $limit, $page), - 'productPriceFilters' => $filters, - 'productPriceStatus' => [], - 'productPriceError' => '商品价格暂时无法读取，请稍后重试。', - ]; + return $this->emptyProductPriceContent( + $filters, + $limit, + $page, + '共享业务数据暂时无法读取，请稍后重试。' + ); } } - private function productPricePaginator(array $rows, int $limit, int $page): Paginator + private function emptyProductPriceContent(array $filters, int $limit, int $page, string $error): array { - return new Paginator( - Collection::make($rows)->map(static fn (array $row): object => (object) $row), - $limit, - $page, - [ - 'path' => request()->url(), - 'query' => request()->except('page'), - ], - ); + return [ + 'productPrices' => new LengthAwarePaginator( + [], + 0, + $limit, + $page, + [ + 'path' => request()->url(), + 'query' => request()->except('page'), + ], + ), + 'productPriceFilters' => $filters, + 'productPriceError' => $error, + ]; } /** diff --git a/compose.jiangling.yml b/compose.jiangling.yml index 7fa5af4a..883a5160 100644 --- a/compose.jiangling.yml +++ b/compose.jiangling.yml @@ -16,7 +16,9 @@ services: retries: 12 start_period: 30s - # 独立的财务系统配置数据库。只开放这个容器，正式业务库 caiwu_db 继续保持内网隔离。 + # 团队共享 MariaDB 实例：逻辑上分为系统配置库和业务镜像库。 + # 两个库独立备份、独立表结构，但复用同一套 TLS 入口和团队密码， + # 避免新人再填写第二套连接资料。正式业务库 caiwu_db 继续保持内网隔离。 config_db: image: mariadb:11.4 pull_policy: missing @@ -25,10 +27,10 @@ services: env_file: - ./runtime/config-db.env command: - - "--innodb-buffer-pool-size=64M" - - "--innodb-log-buffer-size=8M" - - "--max-connections=30" - - "--table-open-cache=200" + - "--innodb-buffer-pool-size=512M" + - "--innodb-log-buffer-size=32M" + - "--max-connections=60" + - "--table-open-cache=600" - "--performance-schema=OFF" ports: - "0.0.0.0:13306:3306" @@ -38,8 +40,8 @@ services: networks: - default - config_database - mem_limit: 256m - cpus: "0.50" + mem_limit: 1536m + cpus: "2.00" healthcheck: test: ["CMD", "healthcheck.sh", "--connect", "--innodb_initialized"] interval: 10s diff --git a/config/database.php b/config/database.php index c46cb383..f667adad 100644 --- a/config/database.php +++ b/config/database.php @@ -107,6 +107,42 @@ return [ 'options' => $mysqlSslOptions(), ], + // 上游业务数据的项目内镜像。它与用户、权限和系统设置所在的 + // caiwu_system_config 分库保存，但默认复用同一台 MariaDB 和账号。 + 'business_mirror' => [ + 'driver' => 'mysql', + 'read' => [ + 'host' => [ + env('BUSINESS_MIRROR_DB_HOST_READ', env('BUSINESS_MIRROR_DB_HOST', env('DB_HOST', '127.0.0.1'))), + ], + ], + 'write' => [ + 'host' => [ + env('BUSINESS_MIRROR_DB_HOST_WRITE', env('BUSINESS_MIRROR_DB_HOST', env('DB_HOST', '127.0.0.1'))), + ], + ], + 'sticky' => true, + 'port' => env('BUSINESS_MIRROR_DB_PORT', env('DB_PORT', '3306')), + 'database' => env('BUSINESS_MIRROR_DB_DATABASE', 'caiwu_business_mirror'), + 'username' => env('BUSINESS_MIRROR_DB_USERNAME', env('DB_USERNAME', 'forge')), + 'password' => env('BUSINESS_MIRROR_DB_PASSWORD', env('DB_PASSWORD', '')), + 'unix_socket' => env('BUSINESS_MIRROR_DB_SOCKET', env('DB_SOCKET', '')), + 'charset' => 'utf8mb4', + 'collation' => 'utf8mb4_unicode_ci', + 'prefix' => '', + 'prefix_indexes' => true, + 'strict' => true, + 'engine' => 'InnoDB ROW_FORMAT=DYNAMIC', + 'modes' => [ + 'STRICT_TRANS_TABLES', + 'NO_ZERO_IN_DATE', + 'NO_ZERO_DATE', + 'ERROR_FOR_DIVISION_BY_ZERO', + 'NO_ENGINE_SUBSTITUTION', + ], + 'options' => $mysqlSslOptions(), + ], + 'mariadb' => [ 'driver' => 'mysql', 'url' => env('DATABASE_URL'), diff --git a/config/upstream_mirrors.php b/config/upstream_mirrors.php new file mode 100644 index 00000000..68573f37 --- /dev/null +++ b/config/upstream_mirrors.php @@ -0,0 +1,21 @@ + env('BUSINESS_MIRROR_DB_CONNECTION', 'business_mirror'), + + // 只有生产刷新节点写镜像；开发环境和网页请求只读共享结果。 + 'refresh_enabled' => env( + 'BUSINESS_MIRROR_REFRESH_ENABLED', + env('APP_ENV', 'production') === 'production' + ), + + 'product_prices' => [ + 'dataset_key' => 'supplier-payables.product-prices', + 'source_key' => 'operations_replica', + 'source_object' => 'rpt.Bulk_Price_Limit', + 'batch_size' => (int) env('PRODUCT_PRICE_MIRROR_BATCH_SIZE', 500), + 'max_batches_per_run' => (int) env('PRODUCT_PRICE_MIRROR_MAX_BATCHES', 4), + // 上游完成记录通常不再变化；仍回看 48 小时，覆盖迟到或修正数据。 + 'overlap_hours' => (int) env('PRODUCT_PRICE_MIRROR_OVERLAP_HOURS', 48), + ], +]; diff --git a/database/migrations/2026_08_14_200000_create_product_price_mirror_tables.php b/database/migrations/2026_08_14_200000_create_product_price_mirror_tables.php new file mode 100644 index 00000000..6183818e --- /dev/null +++ b/database/migrations/2026_08_14_200000_create_product_price_mirror_tables.php @@ -0,0 +1,86 @@ +environment('testing') + || (string) config('system_database.mode', 'bootstrap') === 'bootstrap') { + return; + } + + $schema = Schema::connection($this->connection()); + + if (! $schema->hasTable('business_mirror_sync_states')) { + $schema->create('business_mirror_sync_states', function (Blueprint $table): void { + $table->string('dataset_key', 120)->primary(); + $table->string('source_key', 64); + $table->string('source_object', 191); + $table->string('status', 24)->default('pending'); + $table->string('run_mode', 24)->nullable(); + $table->dateTime('window_started_at')->nullable(); + $table->dateTime('window_ends_at')->nullable(); + // SQL Server datetime2(7) 比 PHP/MySQL 常用的 6 位小数更精细， + // 游标必须按原始 ISO 字符串保存，不能截断后造成重复或漏行。 + $table->string('cursor_completed_at', 33)->nullable(); + $table->string('cursor_business_id', 191)->nullable(); + $table->string('cursor_product_code', 191)->nullable(); + $table->dateTime('initial_completed_at')->nullable(); + $table->dateTime('last_attempted_at')->nullable(); + $table->dateTime('last_succeeded_at')->nullable(); + $table->unsignedBigInteger('last_upserted_rows')->default(0); + $table->string('last_error_code', 80)->nullable(); + $table->timestamps(); + }); + } + + if (! $schema->hasTable('product_price_mirror_rows')) { + $schema->create('product_price_mirror_rows', function (Blueprint $table): void { + $table->bigIncrements('id'); + $table->char('upstream_key', 64)->unique(); + $table->string('business_id', 191); + $table->string('barcode', 191)->index(); + $table->string('spu', 191)->nullable()->index(); + $table->text('product_name')->nullable(); + $table->string('category', 255)->nullable()->index(); + $table->string('production_status', 120)->nullable(); + $table->decimal('minimum_price', 20, 4)->nullable(); + $table->decimal('price_limit', 20, 4)->nullable(); + $table->decimal('purchase_price', 20, 4)->nullable(); + $table->decimal('shipping_fee', 20, 4)->nullable(); + $table->decimal('jingxi_limit_price', 20, 4)->nullable(); + $table->string('approval_status', 120)->nullable(); + $table->string('approval_result', 120)->nullable(); + $table->text('workflow_title')->nullable(); + $table->dateTime('completed_at', 6)->index(); + $table->char('row_hash', 64); + $table->dateTime('synced_at'); + $table->timestamps(); + + $table->index(['completed_at', 'business_id', 'barcode'], 'product_price_mirror_order'); + }); + } + } + + public function down(): void + { + if (app()->environment('testing') + || (string) config('system_database.mode', 'bootstrap') === 'bootstrap') { + return; + } + + $schema = Schema::connection($this->connection()); + $schema->dropIfExists('product_price_mirror_rows'); + $schema->dropIfExists('business_mirror_sync_states'); + } +}; diff --git a/deploy/activate-incremental-release.sh b/deploy/activate-incremental-release.sh index f59085a0..9bc68558 100644 --- a/deploy/activate-incremental-release.sh +++ b/deploy/activate-incremental-release.sh @@ -65,6 +65,28 @@ fi cd "$REMOTE_DIR" export IMAGE_TAG RUN_MIGRATIONS + +if [ "$RUN_MIGRATIONS" = "1" ]; then + # 先创建独立业务镜像库并授权应用账号，再启动会执行迁移的新应用。 + # 该脚本幂等；已有数据库只会补齐授权，不会覆盖业务数据。 + docker compose -f compose.jiangling.yml up -d config_db + config_db_attempt=1 + while [ "$config_db_attempt" -le 24 ]; do + if [ "$(docker inspect caiwu_config_db --format '{{.State.Health.Status}}' 2>/dev/null || true)" = "healthy" ]; then + break + fi + + sleep 5 + config_db_attempt=$((config_db_attempt + 1)) + done + if [ "$(docker inspect caiwu_config_db --format '{{.State.Health.Status}}' 2>/dev/null || true)" != "healthy" ]; then + docker compose -f compose.jiangling.yml logs --tail 80 config_db >&2 || true + echo "Shared MariaDB did not become healthy before mirror database initialization" >&2 + exit 1 + fi + docker exec caiwu_config_db sh /docker-entrypoint-initdb.d/10-config-db-security.sh +fi + docker compose -f compose.jiangling.yml up -d --remove-orphans healthy=0 @@ -103,10 +125,10 @@ fi printf '%s\n' "$DEPLOY_SHA" > "$REMOTE_DIR/.deployed-sha.tmp" mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha" -# 只运行已登记来源的共享快照刷新命令，不启用整个 Laravel 调度器， -# 避免意外启动账单提醒等既有任务。共享锁保证多环境不会重复访问上游。 +# 只运行已登记的上游后台读取，不启用整个 Laravel 调度器，避免意外启动 +# 账单提醒等既有任务。同一把锁保证快照与业务镜像不会并发施压上游。 cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF' -* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet >/dev/null 2>&1 +* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet' >/dev/null 2>&1 EOF chmod 644 /etc/cron.d/caiwu-jiansun-snapshots diff --git a/deploy/activate-release.sh b/deploy/activate-release.sh index d6cbb45b..31559d6f 100644 --- a/deploy/activate-release.sh +++ b/deploy/activate-release.sh @@ -69,6 +69,28 @@ fi cd "$REMOTE_DIR" export IMAGE_TAG RUN_MIGRATIONS + +if [ "$RUN_MIGRATIONS" = "1" ]; then + # 先创建独立业务镜像库并授权应用账号，再启动会执行迁移的新应用。 + # 该脚本幂等；已有数据库只会补齐授权，不会覆盖业务数据。 + docker compose -f compose.jiangling.yml up -d config_db + config_db_attempt=1 + while [ "$config_db_attempt" -le 24 ]; do + if [ "$(docker inspect caiwu_config_db --format '{{.State.Health.Status}}' 2>/dev/null || true)" = "healthy" ]; then + break + fi + + sleep 5 + config_db_attempt=$((config_db_attempt + 1)) + done + if [ "$(docker inspect caiwu_config_db --format '{{.State.Health.Status}}' 2>/dev/null || true)" != "healthy" ]; then + docker compose -f compose.jiangling.yml logs --tail 80 config_db >&2 || true + echo "Shared MariaDB did not become healthy before mirror database initialization" >&2 + exit 1 + fi + docker exec caiwu_config_db sh /docker-entrypoint-initdb.d/10-config-db-security.sh +fi + docker compose -f compose.jiangling.yml up -d --remove-orphans healthy=0 @@ -107,10 +129,10 @@ fi printf '%s\n' "$DEPLOY_SHA" > "$REMOTE_DIR/.deployed-sha.tmp" mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha" -# 只运行已登记来源的共享快照刷新命令，不启用整个 Laravel 调度器， -# 避免意外启动账单提醒等既有任务。共享锁保证多环境不会重复访问上游。 +# 只运行已登记的上游后台读取，不启用整个 Laravel 调度器，避免意外启动 +# 账单提醒等既有任务。同一把锁保证快照与业务镜像不会并发施压上游。 cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF' -* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet >/dev/null 2>&1 +* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet' >/dev/null 2>&1 EOF chmod 644 /etc/cron.d/caiwu-jiansun-snapshots diff --git a/deploy/bootstrap-server.sh b/deploy/bootstrap-server.sh index 90245fe9..d924d294 100644 --- a/deploy/bootstrap-server.sh +++ b/deploy/bootstrap-server.sh @@ -44,6 +44,7 @@ if [ ! -f "$REMOTE_DIR/runtime/config-db.env" ]; then { printf '%s\n' \ 'MARIADB_DATABASE=caiwu_system_config' \ + 'BUSINESS_MIRROR_DATABASE=caiwu_business_mirror' \ 'MARIADB_USER=caiwu_config_writer' \ "MARIADB_PASSWORD=$CONFIG_DB_PASSWORD" \ "MARIADB_ROOT_PASSWORD=$CONFIG_DB_ROOT_PASSWORD" \ diff --git a/deploy/config-db-init.sh b/deploy/config-db-init.sh index bce1e4b5..4701d156 100755 --- a/deploy/config-db-init.sh +++ b/deploy/config-db-init.sh @@ -8,6 +8,19 @@ case "${MARIADB_USER:-}" in ;; esac +BUSINESS_MIRROR_DATABASE="${BUSINESS_MIRROR_DATABASE:-caiwu_business_mirror}" + +case "$BUSINESS_MIRROR_DATABASE" in + ''|*[!A-Za-z0-9_]* ) + echo 'BUSINESS_MIRROR_DATABASE contains unsupported characters' >&2 + exit 1 + ;; +esac + mariadb --protocol=socket -uroot -p"$MARIADB_ROOT_PASSWORD" < 单来源适配器 -> 只读查询执行器 -> 对应上游 +本地探索/开发验证 -> 单来源适配器 -> 只读查询执行器 -> 对应上游 -生产页面/API -> 共享查询快照 -> 后台刷新器 -> 只读查询执行器 -> 对应上游 - | - +-> 受控物化到共享标准表/快照 -> 跨来源报表 +生产同步节点 -> 稳定游标分批读取 -> caiwu_business_mirror + +本地与生产页面/API -> caiwu_business_mirror -> 本地分页、筛选、汇总 + +尚未迁移的旧页面 -> 共享查询快照（过渡兼容，不再扩展） ``` 执行规则： -1. 页面只能通过 `UpstreamQueryCache` 或现有兼容入口读取，不能直接连接上游； +1. 已接入镜像的页面只能通过镜像 Repository 读取，不能直接连接上游；尚未迁移的旧页面暂时通过 `UpstreamQueryCache` 兼容； 2. 新适配器只能通过 `UpstreamReadOnlyQuery` 执行固定、参数化的 `SELECT/WITH`，不能把 PDO 交给控制器，不能执行写入、建表、存储过程或多语句； -3. 生产快照键必须包含来源、数据范围、查询条件以及公司/权限范围，不能按“登录账号个数”复制缓存，也不能让不同权限共用结果； +3. 镜像表必须保留来源标识、稳定上游业务键、同步时间和必要行哈希；列表分页与筛选只在共享业务库执行，不能为每个页码复制一份上游结果； 4. 生产刷新每个来源默认最多 2 个冷查询并发、每分钟 6 个查询组、每轮刷新 2 个活跃查询；所有来源共用默认 4 个总槽位； 5. 生产环境中一个来源断线或熔断只影响自身；本地直连失败时直接返回脱敏错误，不阻塞另一个来源； -6. 跨来源报表禁止在 HTTP 请求中实时 JOIN 多个数据库。先分别同步到共享快照或标准表并记录 `source_key`、上游业务键和同步时间，再从共享数据生成报表； +6. 跨来源报表禁止在 HTTP 请求中实时 JOIN 多个数据库。先分别同步到共享业务镜像并记录 `source_key`、上游业务键和同步时间，再从共享数据生成报表； 7. 新来源必须先完成单表、单查询探测并登记 `UpstreamSnapshotRefresher`，最后才允许开启；仅写入 `config/upstreams.php` 不代表已经接入完成； 8. 当前登记的后续来源模板保持关闭，尚未完成表到业务页面的映射和刷新器前，不得填入真实配置或开启总开关。 @@ -203,24 +207,47 @@ UPSTREAM_SLOT_WAIT_MS=1500 php artisan upstream:snapshots:refresh ``` -生产环境使用下面的总入口生成共享快照。本地开发页面直接读取上游，不依赖该任务： +尚未迁移页面仍由下面的兼容入口生成查询快照： ```bash php artisan upstream:snapshots:refresh-all ``` -旧的 `php artisan jiansun:snapshots:refresh` 继续保留兼容。新增来源只需登记自己的 `UpstreamSnapshotRefresher`，不再为每台开发机单独添加 cron。 +旧的 `php artisan jiansun:snapshots:refresh` 继续保留兼容。已迁移的数据集使用自己的 `business-mirror:sync-*` 命令；部署脚本在同一把上游互斥锁内顺序调用，避免快照刷新和表级同步同时施压上游。 -### 8.2 开发者需要尚未缓存的上游数据时 +### 8.2 共享业务镜像接入闭环 -系统数据库保存团队配置、数据字典和生产快照，不是 SQL Server 网络代理。开发环境和生产环境的读取方式不同： +项目不会复制上游所有表，只复制已经被正式业务代码使用的表或视图。每个数据集必须完成以下闭环： + +1. 明确来源、对象、字段、业务键、稳定排序、增量依据和页面用途； +2. 首次全量同步使用固定时间窗、稳定游标、小批量和断点状态，任何中断都从已提交游标继续； +3. 首次全量完成前页面明确提示同步未完成，不能把第一页、局部快照或空表当成完整数据； +4. 后续同步优先按更新时间、完成时间或递增键增量读取，并保留有限重叠窗口覆盖迟到修正； +5. 上游异常时保留上一版镜像，页面继续读取旧数据并展示最近同步时间；禁止页面现场直连上游补页； +6. 验收至少对比镜像总数、第一页、第二页或末页、指定筛选、最近更新时间和上游限定计数； +7. 没有可靠增量依据的大表必须单独评估低频分段核对，不能在每分钟任务里全表扫描。 + +首个落地数据集是商品价格：上游 `operations_replica / rpt.Bulk_Price_Limit`，镜像表 `product_price_mirror_rows`，同步命令： + +```bash +php artisan business-mirror:sync-product-prices --full +php artisan business-mirror:sync-product-prices +``` + +首次 `--full` 属于生产数据写入，必须先备份 `caiwu_system_config` 和 `caiwu_business_mirror`，并由负责人明确授权。后续无参数命令只执行或继续受控增量。 + +### 8.3 开发者需要尚未接入的上游数据时 + +系统数据库保存团队配置、数据字典和同步状态，共享业务库保存已接入业务行；二者都不是 SQL Server 网络代理。开发探索与正式页面的读取方式不同： ```text 本地开发探索 -> 通用只读入口 -> 使用共享凭据直接读取选定业务表 本地功能实现 -> 固定 Repository -> 使用共享凭据直接只读上游 -生产环境 -> 登记 scope + 参数 -> 后台固定刷新器 -> 共享查询快照 +正式页面 -> 镜像 Repository -> caiwu_business_mirror + +生产同步节点 -> 固定批次 Repository -> 对应上游 ``` 开发者需要先确认表名和字段时，进入“系统设置 → 数据库管理 → 查看数据字典”。任何能直连上游的开发环境或生产刷新节点都可以低频读取两套上游的 `INFORMATION_SCHEMA`，把表、视图、字段、类型和可空信息保存到共享系统数据库 `upstream_schema_columns`。目录同步失败时保留最近一次完整结果，不会清空其他来源。需要更新目录时由 Codex 自动执行： @@ -240,16 +267,16 @@ php artisan upstream:schema:sync operations_replica 1. **先声明边界**：任务和 PR 必须写明 `source_key`、业务用途、表或视图、字段、过滤条件、排序、最大返回量、公司/权限范围、期望 TTL 和没有数据时的页面表现。禁止只写“同步整张表”； 2. **实现固定查询**：Repository/适配器通过 `UpstreamReadOnlyQuery` 执行参数化的 `SELECT/WITH`；控制器和页面只调用 Repository，不接触 PDO，不把 SQL 保存到系统数据库； -3. **登记读取契约**：为列表、汇总和选项定义稳定且可审查的 `scope`，通过 `UpstreamQueryCache` 读取，并在对应 `UpstreamSnapshotRefresher` 中登记同名范围。参数必须有限制，列表必须有稳定排序和行数上限； +3. **登记同步契约**：定义稳定且可审查的数据集键、上游业务键、同步游标、重叠窗口和镜像表。参数必须有限制，批次必须有稳定排序和行数上限； 4. **分支内完成自动测试**：使用脱敏 fixture 或测试替身验证正常、空数据和不可用状态；开发机能够直连时，通过通用只读入口或新 Repository 用少量真实只读数据验证，不得在测试、日志或截图中输出凭据和敏感业务行； -5. **本地直接验证**：本地页面/API 直接执行固定只读查询，不需要配置 `allowed_scopes` 或等待首份快照； -6. **生产登记范围**：代码审查和测试通过后合并到 `main`；生产刷新器仍登记精确 `allowed_scopes` 并生成共享快照； -7. **完成双实例验收**：确认首份快照有同步时间，页面响应头为 `fresh` 或允许的 `stale`；随后从第二个应用实例用相同访问范围读取，并证明 resolver 没有再次访问上游； +5. **本地直接验证**：开发阶段可以直接执行固定只读查询确认数据；正式页面测试使用隔离镜像库 fixture，不依赖负责人环境； +6. **生产首次同步**：代码审查和测试通过后合并到 `main`；经数据库写入授权后创建镜像表并执行可断点的首次全量； +7. **完成共享验收**：确认同步状态成功、镜像总数和筛选计数与上游限定查询一致，第二页或末页可读；随后从第二个应用实例读取相同镜像，并证明页面没有再次访问上游； 8. **发现压力立即收口**：慢查询、返回量异常或上游告警时，先从共享记录移除该精确范围并保留证据，再缩小字段、时间窗或分页。禁止通过提高并发、扩大 `*` 或全表扫描解决。 -部署脚本已经统一安装 `upstream:snapshots:refresh-all`，新增业务查询通常不需要新增 cron 或修改服务器 Compose。发布包含新范围时，发布记录必须额外写明：对应刷新器是否在镜像中、开放了哪些精确范围、首次快照是否成功、第二实例是否复用以及是否观察到慢查询/熔断。未完成这些验收时，范围保持关闭。 +部署脚本在统一上游互斥锁内运行兼容快照刷新和已登记的业务镜像同步。发布新增镜像数据集时，发布记录必须额外写明：镜像表迁移、批次 Repository、首次同步状态、镜像/上游对账、第二页或末页验证、第二实例复用，以及是否观察到慢查询或上游压力。未完成首次全量时，页面必须保持“同步未完成”，不能显示误导性空数据。 -### 8.3 JianSun 现有来源 +### 8.4 旧查询快照兼容范围 该来源按不可修改的只读旧系统处理，不要求对方修改账号、表结构、索引、字段或数据库功能。连接启用状态保存在系统数据库；`JIANSUN_ENABLED` 只用于明确开启的本机覆盖模式。 @@ -322,9 +349,9 @@ php artisan system-database:reencrypt-upstream-passwords --apply 标准共享模式在系统数据库云数据库记录的 `allowed_scopes` 字段中填写逗号分隔的精确范围或 `prefix.*`；本机覆盖模式才使用 `JIANSUN_ALLOWED_SCOPES`。首次上线先开放完成探测的小表范围，例如 `invoice-deductions.*`；约 280 万行核心表对应范围必须单独验证，任何模式下都不得填写全局 `*`。 -生产策略提供的是按代码实际使用范围生成的近实时查询快照，不等同于复制上游全库。本地开发直接读取代码实际调用的数据。没有业务代码调用的数据范围不会查询或同步；禁止为了追求秒级实时而周期性扫描数百万行。 +旧查询快照只用于尚未完成表级迁移的页面，不等同于复制上游全库，也不得继续作为新功能默认方案。没有业务代码调用的表不会同步；新增正式页面按 8.2 接入共享业务镜像。禁止为了追求秒级实时而周期性扫描数百万行。 -### 8.4 生产服务器不能直连上游时 +### 8.5 生产服务器不能直连上游时 使用一台指定开发机作为主中继，另一台作为备用中继。它们只转发 TCP 流量，不保存数据库凭据，也不在开发机执行同步查询。数据库凭据加密保存在系统数据库，中继机只保存已收紧 ACL 的端点和隧道配置；应用仍通过同一套快照、限流和熔断逻辑读取数据。 diff --git a/modules/EcommerceFinance/Repositories/ProductPriceMirrorRepository.php b/modules/EcommerceFinance/Repositories/ProductPriceMirrorRepository.php new file mode 100644 index 00000000..39bda7ee --- /dev/null +++ b/modules/EcommerceFinance/Repositories/ProductPriceMirrorRepository.php @@ -0,0 +1,216 @@ +connection = $database->connection( + (string) config('upstream_mirrors.connection', 'business_mirror') + ); + } + + public function ready(): bool + { + $schema = Schema::connection($this->connection->getName()); + + return $schema->hasTable('business_mirror_sync_states') + && $schema->hasTable('product_price_mirror_rows'); + } + + public function initialized(): bool + { + $state = $this->state(); + + return is_array($state) && ($state['initial_completed_at'] ?? null) !== null; + } + + /** @return array|null */ + public function state(): ?array + { + $row = $this->connection->table('business_mirror_sync_states') + ->where('dataset_key', self::DATASET_KEY) + ->first(); + + return $row === null ? null : (array) $row; + } + + public function paginate(array $filters, int $page, int $limit): LengthAwarePaginator + { + $page = max($page, 1); + $limit = min(max($limit, 1), 100); + $query = $this->connection->table('product_price_mirror_rows'); + + if (($filters['barcode'] ?? '') !== '') { + $query->whereRaw("barcode LIKE ? ESCAPE '!'", [$this->contains((string) $filters['barcode'])]); + } + + if (($filters['completed_from'] ?? '') !== '') { + $query->where('completed_at', '>=', (string) $filters['completed_from'] . ' 00:00:00'); + } + + if (($filters['completed_to'] ?? '') !== '') { + $query->where('completed_at', '<', Carbon::parse((string) $filters['completed_to'])->addDay()->startOfDay()); + } + + return $query + ->orderByDesc('completed_at') + ->orderByDesc('business_id') + ->orderBy('barcode') + ->paginate($limit, [ + 'barcode', 'spu', 'product_name', 'category', 'production_status', + 'minimum_price', 'price_limit', 'purchase_price', 'shipping_fee', + 'jingxi_limit_price', 'approval_status', 'approval_result', + 'workflow_title', 'completed_at', + ], 'page', $page); + } + + public function startRun(string $mode, Carbon $since, Carbon $until, bool $resume): array + { + $now = now(); + $values = [ + 'source_key' => (string) config('upstream_mirrors.product_prices.source_key', 'operations_replica'), + 'source_object' => (string) config('upstream_mirrors.product_prices.source_object', 'rpt.Bulk_Price_Limit'), + 'status' => 'running', + 'run_mode' => $mode, + 'last_attempted_at' => $now, + 'last_error_code' => null, + 'updated_at' => $now, + ]; + + if (! $resume) { + $values += [ + 'window_started_at' => $since, + 'window_ends_at' => $until, + 'cursor_completed_at' => null, + 'cursor_business_id' => null, + 'cursor_product_code' => null, + 'last_upserted_rows' => 0, + 'created_at' => $now, + ]; + } + + $query = $this->connection->table('business_mirror_sync_states') + ->where('dataset_key', self::DATASET_KEY); + + if ($query->exists()) { + $query->update($values); + } else { + $this->connection->table('business_mirror_sync_states')->insert( + ['dataset_key' => self::DATASET_KEY, 'created_at' => $now] + $values, + ); + } + + return $this->state() ?? []; + } + + public function storeBatch(array $rows, array $cursor, int $cumulativeRows): void + { + $now = now(); + $mapped = array_map(function (array $row) use ($now): array { + $businessId = (string) ($row['business_id'] ?? ''); + $barcode = (string) ($row['barcode'] ?? ''); + $cursorCompletedAt = (string) ($row['cursor_completed_at'] ?? $row['completed_at'] ?? ''); + $values = [ + 'business_id' => $businessId, + 'barcode' => $barcode, + 'spu' => $row['spu'] ?? null, + 'product_name' => $row['product_name'] ?? null, + 'category' => $row['category'] ?? null, + 'production_status' => $row['production_status'] ?? null, + 'minimum_price' => $row['minimum_price'] ?? null, + 'price_limit' => $row['price_limit'] ?? null, + 'purchase_price' => $row['purchase_price'] ?? null, + 'shipping_fee' => $row['shipping_fee'] ?? null, + 'jingxi_limit_price' => $row['jingxi_limit_price'] ?? null, + 'approval_status' => $row['approval_status'] ?? null, + 'approval_result' => $row['approval_result'] ?? null, + 'workflow_title' => $row['workflow_title'] ?? null, + 'completed_at' => $row['completed_at'], + ]; + + return $values + [ + 'upstream_key' => hash('sha256', $cursorCompletedAt . '|' . $businessId . '|' . $barcode), + 'row_hash' => hash('sha256', json_encode($values, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION)), + 'synced_at' => $now, + 'created_at' => $now, + 'updated_at' => $now, + ]; + }, $rows); + + $this->connection->transaction(function () use ($mapped, $cursor, $cumulativeRows, $now): void { + if ($mapped !== []) { + $this->connection->table('product_price_mirror_rows')->upsert( + $mapped, + ['upstream_key'], + [ + 'business_id', 'barcode', 'spu', 'product_name', 'category', + 'production_status', 'minimum_price', 'price_limit', 'purchase_price', + 'shipping_fee', 'jingxi_limit_price', 'approval_status', 'approval_result', + 'workflow_title', 'completed_at', 'row_hash', 'synced_at', 'updated_at', + ], + ); + } + + $this->connection->table('business_mirror_sync_states') + ->where('dataset_key', self::DATASET_KEY) + ->update([ + 'cursor_completed_at' => $cursor['completed_at'], + 'cursor_business_id' => $cursor['business_id'], + 'cursor_product_code' => $cursor['barcode'], + 'last_upserted_rows' => $cumulativeRows, + 'updated_at' => $now, + ]); + }); + } + + public function complete(string $mode, Carbon $windowEnd, int $rows): void + { + $now = now(); + $values = [ + 'status' => 'succeeded', + 'last_succeeded_at' => $windowEnd, + 'last_upserted_rows' => $rows, + 'cursor_completed_at' => null, + 'cursor_business_id' => null, + 'cursor_product_code' => null, + 'last_error_code' => null, + 'updated_at' => $now, + ]; + + if ($mode === 'full') { + $values['initial_completed_at'] = $now; + } + + $this->connection->table('business_mirror_sync_states') + ->where('dataset_key', self::DATASET_KEY) + ->update($values); + } + + public function fail(string $errorCode): void + { + $this->connection->table('business_mirror_sync_states') + ->where('dataset_key', self::DATASET_KEY) + ->update([ + 'status' => 'failed', + 'last_error_code' => mb_substr($errorCode, 0, 80), + 'updated_at' => now(), + ]); + } + + private function contains(string $value): string + { + return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], trim($value)) . '%'; + } +} diff --git a/modules/EcommerceFinance/Repositories/ProductPriceRepository.php b/modules/EcommerceFinance/Repositories/ProductPriceRepository.php index e83041f4..9b6a30d3 100644 --- a/modules/EcommerceFinance/Repositories/ProductPriceRepository.php +++ b/modules/EcommerceFinance/Repositories/ProductPriceRepository.php @@ -70,6 +70,62 @@ class ProductPriceRepository return ['rows' => array_map([$this, 'normalize'], $this->query->select($sql, $bindings))]; } + /** + * 按稳定游标分批读取已完成的上游记录，只供后台镜像同步使用。 + * + * @param array{completed_at:string,business_id:string,barcode:string}|null $cursor + * @return array> + */ + public function mirrorBatch(?array $cursor, string $since, string $until, int $limit): array + { + $limit = min(max($limit, 1), 2000); + $bindings = [$since, $until]; + $cursorSql = ''; + + if ($cursor !== null) { + $cursorSql = ' AND (' + . '[process_end_date] > CAST(? AS datetime2(7))' + . " OR ([process_end_date] = CAST(? AS datetime2(7)) AND COALESCE([businessId], '') > ?)" + . " OR ([process_end_date] = CAST(? AS datetime2(7)) AND COALESCE([businessId], '') = ?" + . " AND COALESCE([product_code], '') > ?)" + . ')'; + $bindings = array_merge($bindings, [ + $cursor['completed_at'], + $cursor['completed_at'], + $cursor['business_id'], + $cursor['completed_at'], + $cursor['business_id'], + $cursor['barcode'], + ]); + } + + $sql = "SELECT TOP {$limit} " + . "COALESCE([businessId], '') AS business_id, " + . "COALESCE([product_code], '') AS barcode, " + . '[spu] AS spu, ' + . '[product_name] AS product_name, ' + . '[category] AS category, ' + . '[production_status] AS production_status, ' + . '[min_price] AS minimum_price, ' + . '[price_limit] AS price_limit, ' + . '[purchase_price] AS purchase_price, ' + . '[shipping_fee] AS shipping_fee, ' + . '[jingxi_limit_price] AS jingxi_limit_price, ' + . "CASE [status] WHEN 'COMPLETED' THEN N'已完成' ELSE NULLIF([status], '') END AS approval_status, " + . "CASE [result] WHEN 'agree' THEN N'已同意' WHEN 'refuse' THEN N'已拒绝' ELSE NULLIF([result], '') END AS approval_result, " + . '[title] AS workflow_title, ' + . 'CONVERT(varchar(33), [process_end_date], 126) AS cursor_completed_at, ' + . 'CONVERT(varchar(26), [process_end_date], 121) AS completed_at ' + . 'FROM ' . self::TABLE + . ' WHERE [process_end_date] IS NOT NULL' + . ' AND [process_end_date] >= CAST(? AS datetime2(7))' + . ' AND [process_end_date] < CAST(? AS datetime2(7))' + . $cursorSql + . " ORDER BY [process_end_date] ASC, COALESCE([businessId], '') ASC, COALESCE([product_code], '') ASC"; + + return array_map([$this, 'normalize'], $this->query->select($sql, $bindings)); + } + private function contains(string $value): string { return '%' . str_replace(['\\', '%', '_'], ['\\\\', '\\%', '\\_'], trim($value)) . '%'; diff --git a/modules/EcommerceFinance/Services/ProductPriceMirrorSynchronizer.php b/modules/EcommerceFinance/Services/ProductPriceMirrorSynchronizer.php new file mode 100644 index 00000000..425dede6 --- /dev/null +++ b/modules/EcommerceFinance/Services/ProductPriceMirrorSynchronizer.php @@ -0,0 +1,116 @@ +mirror->ready()) { + throw new RuntimeException('business_mirror_schema_missing'); + } + + $state = $this->mirror->state(); + + if (! $full && ($state['initial_completed_at'] ?? null) === null) { + return ['status' => 'requires_full_sync', 'mode' => 'incremental', 'batches' => 0, 'rows' => 0]; + } + + $mode = $full ? 'full' : 'incremental'; + $resume = is_array($state) + && ($state['run_mode'] ?? null) === $mode + && in_array($state['status'] ?? null, ['running', 'failed'], true) + && ($state['window_started_at'] ?? null) !== null + && ($state['window_ends_at'] ?? null) !== null; + + if ($resume) { + $since = Carbon::parse((string) $state['window_started_at']); + $until = Carbon::parse((string) $state['window_ends_at']); + } elseif ($full) { + $since = Carbon::create(1900, 1, 1, 0, 0, 0, config('app.timezone')); + $until = now()->addSecond(); + } else { + $overlapHours = max((int) config('upstream_mirrors.product_prices.overlap_hours', 48), 1); + $since = Carbon::parse((string) $state['last_succeeded_at'])->subHours($overlapHours); + $until = now()->addSecond(); + } + + $state = $this->mirror->startRun($mode, $since, $until, $resume); + $cursor = $this->cursorFromState($state); + $batchSize ??= (int) config('upstream_mirrors.product_prices.batch_size', 500); + $batchSize = min(max($batchSize, 1), 2000); + $maxBatches ??= (int) config('upstream_mirrors.product_prices.max_batches_per_run', 4); + $maxBatches = max($maxBatches, 0); + $batches = 0; + $rows = (int) ($state['last_upserted_rows'] ?? 0); + + try { + while (true) { + $batch = $this->upstream->mirrorBatch( + $cursor, + $since->format('Y-m-d H:i:s'), + $until->format('Y-m-d H:i:s'), + $batchSize, + ); + + if ($batch === []) { + $this->mirror->complete($mode, $until, $rows); + + return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows]; + } + + $last = $batch[array_key_last($batch)]; + $cursor = [ + 'completed_at' => (string) ($last['cursor_completed_at'] ?? $last['completed_at']), + 'business_id' => (string) $last['business_id'], + 'barcode' => (string) $last['barcode'], + ]; + $rows += count($batch); + $batches++; + $this->mirror->storeBatch($batch, $cursor, $rows); + + if (count($batch) < $batchSize) { + $this->mirror->complete($mode, $until, $rows); + + return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows]; + } + + if ($maxBatches > 0 && $batches >= $maxBatches) { + return ['status' => 'running', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows]; + } + } + } catch (Throwable $exception) { + $this->mirror->fail($exception::class); + throw $exception; + } + } + + /** @return array{completed_at:string,business_id:string,barcode:string}|null */ + private function cursorFromState(array $state): ?array + { + if (($state['cursor_completed_at'] ?? null) === null) { + return null; + } + + return [ + // 直接复用 SQL Server 返回的 datetime2(7) ISO 字符串；Carbon 只有 + // 6 位微秒精度，解析再格式化会丢掉最后一位并破坏稳定游标。 + 'completed_at' => (string) $state['cursor_completed_at'], + 'business_id' => (string) ($state['cursor_business_id'] ?? ''), + 'barcode' => (string) ($state['cursor_product_code'] ?? ''), + ]; + } +} diff --git a/tests/Feature/Business/ProductPriceMirrorRepositoryTest.php b/tests/Feature/Business/ProductPriceMirrorRepositoryTest.php new file mode 100644 index 00000000..615f733f --- /dev/null +++ b/tests/Feature/Business/ProductPriceMirrorRepositoryTest.php @@ -0,0 +1,128 @@ + [ + 'driver' => 'sqlite', + 'database' => ':memory:', + 'prefix' => '', + 'foreign_key_constraints' => true, + ], + 'upstream_mirrors.connection' => 'business_mirror', + ]); + DB::purge('business_mirror'); + $schema = DB::connection('business_mirror')->getSchemaBuilder(); + $schema->create('business_mirror_sync_states', function (Blueprint $table): void { + $table->string('dataset_key')->primary(); + $table->string('source_key')->nullable(); + $table->string('source_object')->nullable(); + $table->string('status')->nullable(); + $table->string('run_mode')->nullable(); + $table->string('window_started_at')->nullable(); + $table->string('window_ends_at')->nullable(); + $table->string('cursor_completed_at')->nullable(); + $table->string('cursor_business_id')->nullable(); + $table->string('cursor_product_code')->nullable(); + $table->dateTime('initial_completed_at')->nullable(); + $table->dateTime('last_attempted_at')->nullable(); + $table->dateTime('last_succeeded_at')->nullable(); + $table->unsignedBigInteger('last_upserted_rows')->default(0); + $table->string('last_error_code')->nullable(); + $table->timestamps(); + }); + $schema->create('product_price_mirror_rows', function (Blueprint $table): void { + $table->id(); + $table->string('upstream_key')->unique(); + $table->string('barcode'); + $table->string('spu')->nullable(); + $table->text('product_name')->nullable(); + $table->string('category')->nullable(); + $table->string('production_status')->nullable(); + $table->decimal('minimum_price')->nullable(); + $table->decimal('price_limit')->nullable(); + $table->decimal('purchase_price')->nullable(); + $table->decimal('shipping_fee')->nullable(); + $table->decimal('jingxi_limit_price')->nullable(); + $table->string('approval_status')->nullable(); + $table->string('approval_result')->nullable(); + $table->text('workflow_title')->nullable(); + $table->dateTime('completed_at'); + $table->string('business_id'); + $table->string('row_hash'); + $table->dateTime('synced_at'); + $table->timestamps(); + }); + } + + public function test_page_two_and_literal_wildcard_filter_are_read_from_shared_mirror(): void + { + DB::connection('business_mirror')->table('business_mirror_sync_states')->insert([ + 'dataset_key' => ProductPriceMirrorRepository::DATASET_KEY, + 'initial_completed_at' => '2026-08-14 10:00:00', + ]); + + foreach (range(1, 30) as $index) { + DB::connection('business_mirror')->table('product_price_mirror_rows')->insert([ + 'barcode' => $index === 30 ? 'SKU%_LITERAL' : sprintf('SKU%03d', $index), + 'business_id' => sprintf('B%03d', $index), + 'completed_at' => sprintf('2026-08-%02d 10:00:00', min($index, 28)), + 'upstream_key' => hash('sha256', (string) $index), + 'row_hash' => hash('sha256', 'row-' . $index), + 'synced_at' => '2026-08-14 10:00:00', + ]); + } + + $repository = app(ProductPriceMirrorRepository::class); + $pageTwo = $repository->paginate([], 2, 25); + $literal = $repository->paginate(['barcode' => '%_'], 1, 25); + + $this->assertTrue($repository->initialized()); + $this->assertSame(30, $pageTwo->total()); + $this->assertCount(5, $pageTwo->items()); + $this->assertSame(1, $literal->total()); + $this->assertSame('SKU%_LITERAL', $literal->items()[0]->barcode); + } + + public function test_same_business_and_barcode_at_different_completion_times_are_preserved(): void + { + DB::connection('business_mirror')->table('business_mirror_sync_states')->insert([ + 'dataset_key' => ProductPriceMirrorRepository::DATASET_KEY, + 'initial_completed_at' => '2026-08-14 10:00:00', + ]); + $repository = app(ProductPriceMirrorRepository::class); + $rows = [[ + 'business_id' => 'B100', + 'barcode' => 'SKU100', + 'cursor_completed_at' => '2026-08-14T10:00:00.1234567', + 'completed_at' => '2026-08-14 10:00:00.123456', + ], [ + 'business_id' => 'B100', + 'barcode' => 'SKU100', + 'cursor_completed_at' => '2026-08-14T11:00:00.1234567', + 'completed_at' => '2026-08-14 11:00:00.123456', + ]]; + + $repository->storeBatch($rows, [ + 'completed_at' => '2026-08-14T11:00:00.1234567', + 'business_id' => 'B100', + 'barcode' => 'SKU100', + ], 2); + + $this->assertSame(2, DB::connection('business_mirror')->table('product_price_mirror_rows')->count()); + $state = $repository->state(); + $this->assertSame('2026-08-14T11:00:00.1234567', $state['cursor_completed_at']); + $this->assertSame(2, (int) $state['last_upserted_rows']); + } +} diff --git a/tests/Feature/Business/SupplierPayablePagesTest.php b/tests/Feature/Business/SupplierPayablePagesTest.php index 314cbf98..9178fe3a 100644 --- a/tests/Feature/Business/SupplierPayablePagesTest.php +++ b/tests/Feature/Business/SupplierPayablePagesTest.php @@ -3,6 +3,8 @@ namespace Tests\Feature\Business; use App\Models\Auth\Permission; +use Illuminate\Pagination\LengthAwarePaginator; +use Modules\EcommerceFinance\Repositories\ProductPriceMirrorRepository; use Modules\EcommerceFinance\Services\JianSunQueryCache; use Modules\EcommerceFinance\Services\UpstreamQueryCache; use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory; @@ -111,16 +113,19 @@ class SupplierPayablePagesTest extends FeatureTestCase { $this->grantAdminPermission(); - $cache = \Mockery::mock(UpstreamQueryCache::class); - $cache->shouldReceive('remember') + $repository = \Mockery::mock(ProductPriceMirrorRepository::class); + $repository->shouldReceive('ready')->once()->andReturnTrue(); + $repository->shouldReceive('initialized')->once()->andReturnTrue(); + $repository->shouldReceive('paginate') ->once() - ->withArgs(static function (string $scope, array $parameters): bool { - return $scope === 'supplier-payables.product-prices.list' - && $parameters['filters']['barcode'] === 'YMJ005203N1176V01' - && $parameters['filters']['completed_from'] === '2025-12-01' - && $parameters['filters']['completed_to'] === '2025-12-31'; + ->withArgs(static function (array $filters, int $page, int $limit): bool { + return $filters === [ + 'barcode' => 'YMJ005203N1176V01', + 'completed_from' => '2025-12-01', + 'completed_to' => '2025-12-31', + ] && $page === 1 && $limit > 0; }) - ->andReturn(['rows' => [[ + ->andReturn(new LengthAwarePaginator([(object) [ 'barcode' => 'YMJ005203N1176V01', 'spu' => 'YMJ0052', 'product_name' => '款式二 36/36/176 胡桃木色 BY-008', @@ -135,14 +140,8 @@ class SupplierPayablePagesTest extends FeatureTestCase 'approval_result' => '已同意', 'workflow_title' => '特殊限价需求申请', 'completed_at' => '2025-12-31 00:00:00', - ]]]); - $cache->shouldReceive('responseHeaders')->once()->andReturn([]); - $factory = \Mockery::mock(UpstreamQueryCacheFactory::class); - $factory->shouldReceive('forSource') - ->once() - ->with('operations_replica') - ->andReturn($cache); - $this->app->instance(UpstreamQueryCacheFactory::class, $factory); + ] ], 1, 25, 1)); + $this->app->instance(ProductPriceMirrorRepository::class, $repository); $response = $this->loginAs()->get(route('business.pages.show', [ 'section' => 'supplier-payables', @@ -169,6 +168,29 @@ class SupplierPayablePagesTest extends FeatureTestCase $this->assertCount(15, $headers[0]); } + public function testProductPricesPageDoesNotTreatUninitializedMirrorAsEmptyData(): void + { + $this->grantAdminPermission(); + + $repository = \Mockery::mock(ProductPriceMirrorRepository::class); + $repository->shouldReceive('ready')->once()->andReturnTrue(); + $repository->shouldReceive('initialized')->once()->andReturnFalse(); + $repository->shouldNotReceive('paginate'); + $this->app->instance(ProductPriceMirrorRepository::class, $repository); + + $url = route('business.pages.show', [ + 'company_id' => $this->company->id, + 'section' => 'supplier-payables', + 'page' => 'product-prices', + ]) . '?page=2'; + $response = $this->loginAs()->get($url); + + $response + ->assertOk() + ->assertSee('商品价格正在执行首次全量同步') + ->assertDontSee('没有符合当前筛选条件的商品价格数据'); + } + public function testCompensationPageUsesTheFiveColumnsFromTheReferenceSheet(): void { $this->grantAdminPermission(); diff --git a/tests/Unit/EcommerceFinance/ProductPriceMirrorSynchronizerTest.php b/tests/Unit/EcommerceFinance/ProductPriceMirrorSynchronizerTest.php new file mode 100644 index 00000000..f539f8e6 --- /dev/null +++ b/tests/Unit/EcommerceFinance/ProductPriceMirrorSynchronizerTest.php @@ -0,0 +1,75 @@ +createMock(ProductPriceRepository::class); + $upstream->expects($this->never())->method('mirrorBatch'); + $mirror = $this->createMock(ProductPriceMirrorRepository::class); + $mirror->method('ready')->willReturn(true); + $mirror->method('state')->willReturn(null); + + $result = (new ProductPriceMirrorSynchronizer($upstream, $mirror))->sync(); + + $this->assertSame('requires_full_sync', $result['status']); + } + + public function test_full_sync_upserts_batches_and_marks_initial_sync_complete(): void + { + Carbon::setTestNow('2026-08-14 15:00:00'); + $rows = [[ + 'business_id' => 'B1', + 'barcode' => 'SKU1', + 'cursor_completed_at' => '2026-08-01T10:00:00.1234567', + 'completed_at' => '2026-08-01 10:00:00', + ], [ + 'business_id' => 'B2', + 'barcode' => 'SKU2', + 'cursor_completed_at' => '2026-08-01T11:00:00.7654321', + 'completed_at' => '2026-08-01 11:00:00', + ]]; + + $upstream = $this->createMock(ProductPriceRepository::class); + $upstream->expects($this->exactly(2))->method('mirrorBatch') + ->willReturnOnConsecutiveCalls($rows, []); + + $mirror = $this->createMock(ProductPriceMirrorRepository::class); + $mirror->method('ready')->willReturn(true); + $mirror->method('state')->willReturn(null); + $mirror->expects($this->once())->method('startRun') + ->with('full', $this->isInstanceOf(Carbon::class), $this->isInstanceOf(Carbon::class), false) + ->willReturn(['last_upserted_rows' => 0]); + $mirror->expects($this->once())->method('storeBatch') + ->with($rows, [ + 'completed_at' => '2026-08-01T11:00:00.7654321', + 'business_id' => 'B2', + 'barcode' => 'SKU2', + ], 2); + $mirror->expects($this->once())->method('complete') + ->with('full', $this->isInstanceOf(Carbon::class), 2); + + $result = (new ProductPriceMirrorSynchronizer($upstream, $mirror))->sync(true, 2, 0); + + $this->assertSame([ + 'status' => 'succeeded', + 'mode' => 'full', + 'batches' => 1, + 'rows' => 2, + ], $result); + } +} diff --git a/tests/Unit/EcommerceFinance/ProductPriceRepositoryTest.php b/tests/Unit/EcommerceFinance/ProductPriceRepositoryTest.php index 66a7abac..0e37017a 100644 --- a/tests/Unit/EcommerceFinance/ProductPriceRepositoryTest.php +++ b/tests/Unit/EcommerceFinance/ProductPriceRepositoryTest.php @@ -70,4 +70,51 @@ class ProductPriceRepositoryTest extends TestCase $this->assertNull($result['rows'][1]['minimum_price']); $this->assertSame(0.0, $result['rows'][1]['purchase_price']); } + + public function test_mirror_batch_uses_stable_keyset_cursor_and_bounded_window(): void + { + $query = $this->createMock(UpstreamReadOnlyQuery::class); + $query->expects($this->once())->method('select')->willReturnCallback( + function (string $sql, array $bindings): array { + UpstreamReadOnlyQuery::assertReadOnlySql($sql); + $this->assertStringContainsString('SELECT TOP 200', $sql); + $this->assertStringContainsString("COALESCE([businessId], '') AS business_id", $sql); + $this->assertStringContainsString('[process_end_date] > CAST(? AS datetime2(7))', $sql); + $this->assertStringContainsString('CONVERT(varchar(33), [process_end_date], 126) AS cursor_completed_at', $sql); + $this->assertStringNotContainsString('OFFSET', $sql); + $this->assertSame([ + '2026-08-01 00:00:00', + '2026-08-14 00:00:00', + '2026-08-10 12:00:00', + '2026-08-10 12:00:00', + 'B100', + '2026-08-10 12:00:00', + 'B100', + 'SKU100', + ], $bindings); + + return [[ + 'business_id' => 'B101', + 'barcode' => 'SKU101', + 'completed_at' => '2026-08-10 12:01:00', + 'minimum_price' => '10.25', + 'price_limit' => null, + 'purchase_price' => '8.00', + 'shipping_fee' => '1.00', + 'jingxi_limit_price' => '9.00', + ]]; + }, + ); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($query); + + $rows = (new ProductPriceRepository($factory))->mirrorBatch([ + 'completed_at' => '2026-08-10 12:00:00', + 'business_id' => 'B100', + 'barcode' => 'SKU100', + ], '2026-08-01 00:00:00', '2026-08-14 00:00:00', 200); + + $this->assertSame('B101', $rows[0]['business_id']); + $this->assertSame(10.25, $rows[0]['minimum_price']); + } }