diff --git a/app/Console/Commands/WarmReportCaches.php b/app/Console/Commands/WarmReportCaches.php new file mode 100644 index 00000000..ce1bddfe --- /dev/null +++ b/app/Console/Commands/WarmReportCaches.php @@ -0,0 +1,49 @@ +startOfMonth()->subMonth()->format('Y-m'); + $result = $inventory->paginateMonthlyInventoryReport([ + 'month' => $month, + 'product_code' => '', + 'warehouse_provider' => '', + 'category' => '', + ], 1, 25); + $this->info(sprintf('进销存 %s 首页已预热,共 %d 行。', $month, $result['total'])); + } catch (Throwable $exception) { + $failed = true; + Log::error('默认进销存报表缓存预热失败。', ['exception' => $exception::class]); + $this->error('默认进销存报表缓存预热失败。'); + } + + try { + $result = $alipay->paginate([], 'occurred_at', 'desc', 1, 10); + $this->info(sprintf('淘系支付宝首页已预热,共 %d 行。', $result['total'])); + } catch (Throwable $exception) { + $failed = true; + Log::error('淘系支付宝报表缓存预热失败。', ['exception' => $exception::class]); + $this->error('淘系支付宝报表缓存预热失败。'); + } + + return $failed ? self::FAILURE : self::SUCCESS; + } +} diff --git a/compose.jiangling.yml b/compose.jiangling.yml index 31d22d45..c2ce3e39 100644 --- a/compose.jiangling.yml +++ b/compose.jiangling.yml @@ -27,11 +27,16 @@ services: env_file: - ./runtime/config-db.env command: - - "--innodb-buffer-pool-size=512M" - - "--innodb-log-buffer-size=32M" + - "--innodb-buffer-pool-size=4G" + - "--innodb-log-buffer-size=64M" - "--max-connections=60" - "--table-open-cache=600" - "--performance-schema=OFF" + - "--slow-query-log=ON" + - "--long-query-time=5" + - "--min-examined-row-limit=10000" + - "--log-output=FILE" + - "--slow-query-log-file=/var/lib/mysql/caiwu-slow.log" ports: - "0.0.0.0:13306:3306" volumes: @@ -40,8 +45,8 @@ services: networks: - default - config_database - mem_limit: 1536m - cpus: "2.00" + mem_limit: 6g + cpus: "4.00" healthcheck: test: ["CMD", "healthcheck.sh", "--connect", "--innodb_initialized"] interval: 10s diff --git a/deploy/activate-incremental-release.sh b/deploy/activate-incremental-release.sh index 1c24071e..4f0b2c98 100644 --- a/deploy/activate-incremental-release.sh +++ b/deploy/activate-incremental-release.sh @@ -178,8 +178,10 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha" # 普通商品优先高频增量,并在配置的每周低峰窗口自动继续全量校准。 cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF' CRON_TZ=Asia/Shanghai -* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-store-report --quiet; /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; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1 -0 0 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-third-party-warehouse-storage --max-batches=0 --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 business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /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; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1 +# 店铺经营报表按 15 分钟刷新;单轮覆盖 50000 行回看窗口,避免空闲时持续重读千万级账单。 +7,22,37,52 * * * * root /usr/bin/flock -w 600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-store-report --max-batches=12 --quiet >/dev/null 2>&1 +0 0 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-third-party-warehouse-storage --max-batches=0 --quiet && /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:warm-report-caches --quiet' >/dev/null 2>&1 0 0 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-jushuitan-cost-price-history --quiet >/dev/null 2>&1 0 5 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables operations_replica.rpt.jst_Orders --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-ding-online-daifa-price --quiet' >/dev/null 2>&1 */2 * * * * root /usr/bin/flock -n /run/caiwu-esign-polling.lock /usr/bin/docker exec -u www-data caiwu_app php artisan esign:poll-signatures --quiet >/dev/null 2>&1 diff --git a/deploy/activate-release.sh b/deploy/activate-release.sh index 9cf0655c..b8df34ae 100644 --- a/deploy/activate-release.sh +++ b/deploy/activate-release.sh @@ -182,8 +182,10 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha" # 普通商品优先高频增量,并在配置的每周低峰窗口自动继续全量校准。 cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF' CRON_TZ=Asia/Shanghai -* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-store-report --quiet; /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; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1 -0 0 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-third-party-warehouse-storage --max-batches=0 --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 business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /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; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1 +# 店铺经营报表按 15 分钟刷新;单轮覆盖 50000 行回看窗口,避免空闲时持续重读千万级账单。 +7,22,37,52 * * * * root /usr/bin/flock -w 600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-store-report --max-batches=12 --quiet >/dev/null 2>&1 +0 0 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-third-party-warehouse-storage --max-batches=0 --quiet && /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:warm-report-caches --quiet' >/dev/null 2>&1 0 0 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-jushuitan-cost-price-history --quiet >/dev/null 2>&1 0 5 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables operations_replica.rpt.jst_Orders --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-ding-online-daifa-price --quiet' >/dev/null 2>&1 */2 * * * * root /usr/bin/flock -n /run/caiwu-esign-polling.lock /usr/bin/docker exec -u www-data caiwu_app php artisan esign:poll-signatures --quiet >/dev/null 2>&1 diff --git a/docs/上游业务镜像清单.md b/docs/上游业务镜像清单.md index e5df0d3e..b8472af9 100644 --- a/docs/上游业务镜像清单.md +++ b/docs/上游业务镜像清单.md @@ -96,7 +96,7 @@ 6. 页面切换必须以“总记录数、第一页、第二页、末页、筛选、汇总、空数据和更新时间”作为验收,不能只看首页有数据。 7. 普通商品的采购单同步以采购主表 `updateTime` 加明细 `Id` 作为稳定游标;镜像保存来源单号、供应商、数量、单价和全部状态。普通商品页面按商品编码选择最近完成的 1688 一键代发价格;线上代发价格页面只展示 `Confirmed` / `Finished`,并在共享镜像内关联订单。完成单后续取消时可撤回旧价格;新增投影字段完成新一轮全量前不展示采购单管理数据。 8. 聚水潭官方接口尚未接入时,普通商品历史残留排除由负责人提供的最新完整 XLSX 导出刷新。只排除“导出生成前已创建、但导出中缺失”的镜像行,导出生成后的新增商品保持可见。导入使用新同步代次,只有完整解析成功才原子替换排除表;中断或错误保留上一版。后续接入聚水潭官方“普通商品资料查询”及删除/恢复记录后,由官方接口取代人工导出校准。 -9. 店铺维度报表是已批准的中央专用数据集:生产节点按 `Id` 断点推进并回看最近 50000 行;原始账单同步后增量维护按店铺/日期汇总的 `store_report_daily_rows`,页面不得现场扫描千万级原始账单。开发机不运行刷新 worker。 +9. 店铺维度报表是已批准的中央专用数据集:生产节点每 15 分钟按 `Id` 断点推进并回看最近 50000 行,一轮最多处理 60000 行;同步批次按受影响主键精确维护行数,不得在完成阶段对千万级原始表执行 `COUNT(*)`。原始账单同步后增量维护按店铺/日期汇总的 `store_report_daily_rows`,页面不得现场扫描千万级原始账单。开发机不运行刷新 worker。 10. `operations_replica.rpt.jst_originalBillInfo` 是供多个业务功能复用的原始账单镜像。首次同步由 Codex 备份后自主安排在生产同步节点低峰期分批续传,无需另行授权;页面请求和普通开发机不启动刷新 worker。同步完成后,开发者通过固定 Repository 查询 `ops_jst_original_bill_info`,不得各自创建整表快照。 淘系支付宝账户账单当前仍读取此 MariaDB 镜像,统一调度须持续增量,不因 ClickHouse 尚未启用而暂停。账单冲减则复用“订单数据 → 原始数据 → 订单售后费用明细”,不按该表费用名称推断 500112。 11. 拼多多货款账户账单使用 `jiansun.dbo.jst_originalBillInfo.pinduoduo_receiving` 数据集。同步器必须以参数绑定固定 `channel = PinduoduoImport`,只复制该渠道切片;页面 Repository 只读 `js_jst_pdd_receiving_bill_rows`。首次全量按 `Id` 断点推进,后续按 `updateTime` 保留重叠窗口,不得切回落后一天的 `operations_replica` 副本。 12. 得物完整账单使用聚胜算权威 XLSX,通过 `dewu-bills:import-authoritative` 写入 `dewu_settlement_bill_rows`。导入必须核验文件 SHA-256、账期、记录编码唯一性、行数、收入和支出控制数,并保留得物、退货退款订单、扣减其他费用明细三类渠道;页面 Repository 不得回退 `jiansun.dbo.jst_originalBillInfo.dewu_receiving`。详细流程见 `docs/得物完整账单取数方案留档.md`。 diff --git a/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php b/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php index 40097ba2..843b31e7 100644 --- a/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php @@ -2,6 +2,7 @@ namespace Modules\EcommerceFinance\Repositories; +use Illuminate\Contracts\Cache\LockTimeoutException; use Illuminate\Database\Query\Builder; use Illuminate\Support\Collection; use Illuminate\Support\Facades\Cache; @@ -15,6 +16,12 @@ class StoreReceivingAlipayBillMirrorRepository { private const PAGE_QUERY_TIMEOUT_SECONDS = 30; + private const AGGREGATE_CACHE_SECONDS = 900; + + private const AGGREGATE_CACHE_LOCK_SECONDS = 45; + + private const AGGREGATE_CACHE_WAIT_SECONDS = 35; + /** 淘系支付宝页面不展示其他平台店铺;头条放心购是镜像中的抖音平台名称。 */ private const EXCLUDED_TAOBAO_SHOP_SITE_PREFIXES = [ '拼多多', '京东', '抖音', '头条放心购', '得物', '唯品会', @@ -87,15 +94,13 @@ class StoreReceivingAlipayBillMirrorRepository && $this->channelCoversDataset($channel, $dataset->fixedFilters()) && ! $hasFilters && (($state['status'] ?? null) === 'succeeded' || (bool) ($state['mirror_row_count_is_exact'] ?? false)) && $recordedCount !== null && ctype_digit((string) $recordedCount); - $summary = Cache::remember( + $summary = $this->rememberAggregateSummary( $this->aggregateCacheKey( $datasetKey, $filters, $channel, $excludeNonTaobaoPlatforms, - $state, ), - 120, fn (): array => $this->summary($query, $filters, $channel), ); $total = $useRecordedCount @@ -803,13 +808,12 @@ class StoreReceivingAlipayBillMirrorRepository return (int) ($this->boundedGet($count)->first()->aggregate ?? 0); } - /** @param array $filters @param array $state */ + /** @param array $filters */ private function aggregateCacheKey( string $datasetKey, array $filters, string|array|null $channel, bool $excludeNonTaobaoPlatforms, - array $state, ): string { $normalizedFilters = array_filter( array_map( @@ -824,18 +828,45 @@ class StoreReceivingAlipayBillMirrorRepository sort($normalizedChannel); } - return 'store-receiving-bills:aggregate:v2:' . hash('sha256', (string) json_encode([ + return 'store-receiving-bills:aggregate:v3:' . hash('sha256', (string) json_encode([ 'dataset' => $datasetKey, 'filters' => $normalizedFilters, 'channel' => $normalizedChannel, 'exclude_non_taobao' => $excludeNonTaobaoPlatforms, - 'status' => $state['status'] ?? null, - 'last_succeeded_at' => $state['last_succeeded_at'] ?? null, - 'mirror_row_count' => $state['mirror_row_count'] ?? null, - 'mirror_row_count_is_exact' => $state['mirror_row_count_is_exact'] ?? null, ], JSON_UNESCAPED_UNICODE)); } + /** @param callable():array{income:float,expense:float,count:?int} $resolver */ + private function rememberAggregateSummary(string $key, callable $resolver): array + { + $cached = Cache::get($key); + if (is_array($cached)) { + return $cached; + } + + try { + return Cache::lock($key . ':lock', self::AGGREGATE_CACHE_LOCK_SECONDS) + ->block(self::AGGREGATE_CACHE_WAIT_SECONDS, function () use ($key, $resolver): array { + $cached = Cache::get($key); + if (is_array($cached)) { + return $cached; + } + + $resolved = $resolver(); + Cache::put($key, $resolved, self::AGGREGATE_CACHE_SECONDS); + + return $resolved; + }); + } catch (LockTimeoutException $exception) { + $cached = Cache::get($key); + if (is_array($cached)) { + return $cached; + } + + throw new RuntimeException('store_receiving_bill_aggregate_busy', 0, $exception); + } + } + private function escapeLike(string $value): string { return str_replace(['\\', '%', '_'], ['\\\\', '\\%', '\\_'], $value); diff --git a/modules/EcommerceFinance/Repositories/ThirdPartyWarehouseStorageMirrorRepository.php b/modules/EcommerceFinance/Repositories/ThirdPartyWarehouseStorageMirrorRepository.php index 6094f1d0..10e07960 100644 --- a/modules/EcommerceFinance/Repositories/ThirdPartyWarehouseStorageMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/ThirdPartyWarehouseStorageMirrorRepository.php @@ -4,6 +4,7 @@ namespace Modules\EcommerceFinance\Repositories; use Carbon\CarbonInterface; use Generator; +use Illuminate\Contracts\Cache\LockTimeoutException; use Illuminate\Database\ConnectionInterface; use Illuminate\Database\DatabaseManager; use Illuminate\Database\Query\Builder; @@ -24,6 +25,12 @@ class ThirdPartyWarehouseStorageMirrorRepository { private const REPORT_QUERY_TIMEOUT_SECONDS = 30; + private const REPORT_CACHE_SECONDS = 93600; + + private const REPORT_CACHE_LOCK_SECONDS = 45; + + private const REPORT_CACHE_WAIT_SECONDS = 35; + public const DATASET_KEY = 'jiansun.third-party-warehouse-storage'; public const WAREHOUSE_ID = 15045364; @@ -371,7 +378,7 @@ class ThirdPartyWarehouseStorageMirrorRepository $state, ); $rolledForward = $includeTotal && $perPage <= 100 && $authority !== null - ? Cache::remember( + ? $this->rememberMonthlyInventoryReport( $this->monthlyInventoryReportCacheKey( $period, $filters, @@ -381,13 +388,14 @@ class ThirdPartyWarehouseStorageMirrorRepository $inventoryGeneration, $page, $perPage, - $state, ), - 120, $resolveRollForward, ) : $resolveRollForward(); if ($rolledForward !== null) { + $rolledForward['state'] = $state; + $rolledForward['sync'] = $this->syncStatus($state); + return $rolledForward; } @@ -904,7 +912,6 @@ class ThirdPartyWarehouseStorageMirrorRepository * @param array{start:string,end:string} $period * @param array $filters * @param array{generation:string,source_month:string} $authority - * @param array $state */ private function monthlyInventoryReportCacheKey( array $period, @@ -915,7 +922,6 @@ class ThirdPartyWarehouseStorageMirrorRepository string $inventoryGeneration, int $page, int $perPage, - array $state, ): string { $normalizedFilters = [ 'product_code' => trim((string) ($filters['product_code'] ?? '')), @@ -923,20 +929,51 @@ class ThirdPartyWarehouseStorageMirrorRepository 'category' => trim((string) ($filters['category'] ?? '')), ]; - return 'inventory-report:monthly-page:v2:' . hash('sha256', (string) json_encode([ + return 'inventory-report:monthly-page:v3:' . hash('sha256', (string) json_encode([ 'period' => $period, 'filters' => $normalizedFilters, 'authority' => $authority, 'header_generation' => $headerGeneration, 'item_generation' => $itemGeneration, 'inventory_generation' => $inventoryGeneration, - 'status' => $state['status'] ?? null, - 'last_succeeded_at' => $state['last_succeeded_at'] ?? null, 'page' => $page, 'per_page' => $perPage, ], JSON_UNESCAPED_UNICODE)); } + /** @param callable():?array $resolver */ + private function rememberMonthlyInventoryReport(string $key, callable $resolver): ?array + { + $cached = Cache::get($key); + if (is_array($cached)) { + return $cached; + } + + try { + return Cache::lock($key . ':lock', self::REPORT_CACHE_LOCK_SECONDS) + ->block(self::REPORT_CACHE_WAIT_SECONDS, function () use ($key, $resolver): ?array { + $cached = Cache::get($key); + if (is_array($cached)) { + return $cached; + } + + $resolved = $resolver(); + if (is_array($resolved)) { + Cache::put($key, $resolved, self::REPORT_CACHE_SECONDS); + } + + return $resolved; + }); + } catch (LockTimeoutException $exception) { + $cached = Cache::get($key); + if (is_array($cached)) { + return $cached; + } + + throw new RuntimeException('inventory_report_generation_busy', 0, $exception); + } + } + /** @param array{start:string,end:string} $period */ private function previousMonthlyInventoryAuthority(array $period): ?array { diff --git a/modules/EcommerceFinance/Services/StoreReportMirrorSynchronizer.php b/modules/EcommerceFinance/Services/StoreReportMirrorSynchronizer.php index a09bfa9f..8f240e40 100644 --- a/modules/EcommerceFinance/Services/StoreReportMirrorSynchronizer.php +++ b/modules/EcommerceFinance/Services/StoreReportMirrorSynchronizer.php @@ -119,10 +119,16 @@ class StoreReportMirrorSynchronizer } $mapped = $map($rows, $generation); - $this->storeBatch($dataset, $table, $generation, $nextCursor, $mapped, $initial, $storedRows); + $storedRows = $this->storeBatch( + $dataset, + $table, + $generation, + $nextCursor, + $mapped, + $storedRows, + ); $cursor = $nextCursor; $sourceRows += count($rows); - $storedRows += $initial ? count($mapped['upserts']) : 0; $batches++; if (count($rows) < $batchSize) { @@ -163,12 +169,30 @@ class StoreReportMirrorSynchronizer string $generation, int $cursor, array $mapped, - bool $initial, int $storedRows, - ): void { - $this->connection->transaction(function () use ( - $dataset, $table, $generation, $cursor, $mapped, $initial, $storedRows, - ): void { + ): int { + return $this->connection->transaction(function () use ( + $dataset, $table, $generation, $cursor, $mapped, $storedRows, + ): int { + $upsertIds = []; + foreach ($mapped['upserts'] as $row) { + $upsertIds[(int) $row['source_id']] = true; + } + $deleteIds = array_fill_keys(array_map('intval', $mapped['deletes']), true); + $touchedIds = array_keys($upsertIds + $deleteIds); + $existingIds = []; + foreach (array_chunk($touchedIds, 1000) as $chunk) { + foreach ($this->connection->table($table) + ->where('generation', $generation) + ->whereIn('source_id', $chunk) + ->pluck('source_id') as $sourceId) { + $existingIds[(int) $sourceId] = true; + } + } + $inserted = count(array_diff_key($upsertIds, $existingIds)); + $deleted = count(array_diff_key(array_intersect_key($deleteIds, $existingIds), $upsertIds)); + $nextStoredRows = max($storedRows + $inserted - $deleted, 0); + $dailyDeltas = $table === 'store_report_bill_rows' && $this->dailySummaryReady($generation) ? $this->dailyDeltas($generation, $mapped) : []; @@ -195,9 +219,11 @@ class StoreReportMirrorSynchronizer } $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update([ 'cursor_id' => $cursor, - 'row_count' => $initial ? $storedRows + count($mapped['upserts']) : $storedRows, + 'row_count' => $nextStoredRows, 'updated_at' => now(), ]); + + return $nextStoredRows; }); } @@ -212,18 +238,13 @@ class StoreReportMirrorSynchronizer int $sourceRows, ): array { $now = now(); - $actualRows = (int) $this->connection->table( - $dataset === StoreReportRepository::BILLS_DATASET - ? 'store_report_bill_rows' - : 'store_report_lifecycle_rows', - )->where('generation', $generation)->count(); $values = [ 'status' => 'succeeded', 'run_mode' => null, 'cursor_id' => $cursor, 'last_succeeded_at' => $now, 'last_error_code' => null, - 'row_count' => $actualRows, + 'row_count' => $storedRows, 'updated_at' => $now, ]; if ($mode === 'full') { @@ -235,7 +256,7 @@ class StoreReportMirrorSynchronizer } $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update($values); - return $this->result('succeeded', $mode, $batches, $sourceRows, $actualRows, $cursor); + return $this->result('succeeded', $mode, $batches, $sourceRows, $storedRows, $cursor); } /** @return array{status:string,mode:string,batches:int,source_rows:int,stored_rows:int,cursor_id:int} */ diff --git a/tests/Feature/Console/StoreReportMirrorSyncTest.php b/tests/Feature/Console/StoreReportMirrorSyncTest.php index c3604853..8d54bafe 100644 --- a/tests/Feature/Console/StoreReportMirrorSyncTest.php +++ b/tests/Feature/Console/StoreReportMirrorSyncTest.php @@ -69,7 +69,7 @@ class StoreReportMirrorSyncTest extends TestCase public function test_incremental_sync_reuses_active_generation_and_overlaps_recent_ids(): void { $this->activeState(StoreReportRepository::LIFECYCLES_DATASET, 'lifecycle-live', 20, 1); - $this->activeState(StoreReportRepository::BILLS_DATASET, 'bill-live', 100, 10); + $this->activeState(StoreReportRepository::BILLS_DATASET, 'bill-live', 100, 1); $this->activeState(StoreReportRepository::DAILY_DATASET, 'bill-live', 100, 1); DB::connection('business_mirror')->table('store_report_bill_rows')->insert([ 'generation' => 'bill-live', 'source_id' => 101, 'shop_id' => 88, @@ -104,6 +104,52 @@ class StoreReportMirrorSyncTest extends TestCase ], 'business_mirror'); } + public function test_incremental_sync_tracks_inserts_and_deletes_without_counting_the_fact_table(): void + { + $this->activeState(StoreReportRepository::LIFECYCLES_DATASET, 'lifecycle-live', 20, 0); + $this->activeState(StoreReportRepository::BILLS_DATASET, 'bill-live', 100, 2); + $this->activeState(StoreReportRepository::DAILY_DATASET, 'bill-live', 100, 1); + $db = DB::connection('business_mirror'); + $db->table('store_report_bill_rows')->insert([ + $this->storedBill(101, 10), + $this->storedBill(102, 20), + ]); + $db->table('store_report_daily_rows')->insert([ + 'generation' => 'bill-live', 'store_name' => '店铺A', 'occurred_on' => '2026-08-01', + 'income' => 30, 'expense' => 0, 'updated_at' => now(), + ]); + + $deleted = $this->bill(102, '1', 20); + $deleted['del_flag'] = 1; + $upstream = $this->createMock(StoreReportUpstreamRepository::class); + $upstream->expects($this->once())->method('lifecycleBatch')->with(0, 100)->willReturn([]); + $upstream->expects($this->once())->method('billBatch')->with(50, 100)->willReturn([ + $this->bill(101, '1', 11), + $deleted, + $this->bill(103, '-1', 7), + ]); + + $db->enableQueryLog(); + $result = (new StoreReportMirrorSynchronizer($upstream, app('db')))->sync(100, 2); + $queries = $db->getQueryLog(); + $db->disableQueryLog(); + + $this->assertSame(2, $result['bills']['stored_rows']); + $this->assertDatabaseMissing('store_report_bill_rows', [ + 'generation' => 'bill-live', 'source_id' => 102, + ], 'business_mirror'); + $this->assertDatabaseHas('store_report_bill_rows', [ + 'generation' => 'bill-live', 'source_id' => 103, 'expense' => 7, + ], 'business_mirror'); + $this->assertDatabaseHas('store_report_import_states', [ + 'dataset_key' => StoreReportRepository::BILLS_DATASET, 'row_count' => 2, + ], 'business_mirror'); + $this->assertFalse(collect($queries)->contains( + static fn (array $query): bool => str_contains(strtolower($query['query']), 'count(*)') + && str_contains(strtolower($query['query']), 'store_report_bill_rows'), + )); + } + private function bill(int $id, string $direction, float $amount): array { return [ @@ -115,6 +161,18 @@ class StoreReportMirrorSyncTest extends TestCase ]; } + /** @return array */ + private function storedBill(int $id, float $income): array + { + return [ + 'generation' => 'bill-live', 'source_id' => $id, 'shop_id' => 88, + 'platform' => 'TouTiaoFXGZJ', 'platform_label' => '抖音', 'store_name' => '店铺A', + 'occurred_at' => '2026-08-01 10:00:00', 'income' => $income, 'expense' => 0, + 'bill_item' => '订单收入', 'record_code' => (string) $id, 'source_file' => 'source', + 'row_hash' => 'old-' . $id, 'source_updated_at' => null, 'imported_at' => now(), + ]; + } + private function activeState(string $dataset, string $generation, int $cursor, int $rows): void { DB::connection('business_mirror')->table('store_report_import_states')->insert([ diff --git a/tests/Unit/Deployment/BusinessMirrorScheduleTest.php b/tests/Unit/Deployment/BusinessMirrorScheduleTest.php index 2a4d6839..846b73e5 100644 --- a/tests/Unit/Deployment/BusinessMirrorScheduleTest.php +++ b/tests/Unit/Deployment/BusinessMirrorScheduleTest.php @@ -12,7 +12,7 @@ class BusinessMirrorScheduleTest extends TestCase $contents = file_get_contents(dirname(__DIR__, 3) . '/deploy/' . $script); $catalogPosition = strpos($contents, 'business-mirror:sync-product-catalog --quiet'); $purchasePricePosition = strpos($contents, 'business-mirror:sync-product-purchase-prices --quiet'); - $storeReportPosition = strpos($contents, 'business-mirror:sync-store-report --quiet'); + $storeReportPosition = strpos($contents, 'business-mirror:sync-store-report --max-batches=12 --quiet'); $snapshotPosition = strpos($contents, 'upstream:snapshots:refresh-all --quiet'); $pricePosition = strpos($contents, 'business-mirror:sync-product-prices --quiet'); $genericPosition = strpos($contents, 'business-mirror:sync-tables --quiet'); @@ -25,13 +25,24 @@ class BusinessMirrorScheduleTest extends TestCase $this->assertNotFalse($genericPosition, $script); $this->assertLessThan($snapshotPosition, $catalogPosition, $script); $this->assertLessThan($snapshotPosition, $purchasePricePosition, $script); - $this->assertLessThan($snapshotPosition, $storeReportPosition, $script); $this->assertLessThan($purchasePricePosition, $catalogPosition, $script); $this->assertLessThan($pricePosition, $catalogPosition, $script); $this->assertLessThan($genericPosition, $catalogPosition, $script); $this->assertSame(1, substr_count($contents, 'business-mirror:sync-product-catalog --quiet'), $script); $this->assertSame(2, substr_count($contents, 'business-mirror:sync-product-purchase-prices --quiet'), $script); - $this->assertSame(1, substr_count($contents, 'business-mirror:sync-store-report --quiet'), $script); + $this->assertSame(1, substr_count($contents, 'business-mirror:sync-store-report'), $script); + $this->assertStringContainsString( + '7,22,37,52 * * * * root /usr/bin/flock -w 600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-store-report --max-batches=12 --quiet', + $contents, + $script, + ); + $minuteLines = array_values(array_filter( + preg_split('/\R/', $contents) ?: [], + static fn (string $line): bool => str_starts_with($line, '* * * * * root'), + )); + $minuteLine = $minuteLines[0] ?? null; + $this->assertIsString($minuteLine, $script); + $this->assertStringNotContainsString('business-mirror:sync-store-report', $minuteLine, $script); $this->assertStringContainsString('docker exec -u www-data caiwu_app', $contents, $script); $this->assertStringContainsString('/run/caiwu-upstream-sync.lock', $contents, $script); $this->assertStringContainsString("CRON_TZ=Asia/Shanghai\n", $contents, $script); @@ -52,6 +63,12 @@ class BusinessMirrorScheduleTest extends TestCase $contents, $script, ); + $this->assertStringContainsString( + 'business-mirror:sync-third-party-warehouse-storage --max-batches=0 --quiet && /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:warm-report-caches --quiet', + $contents, + $script, + ); + $this->assertSame(1, substr_count($contents, 'business-mirror:warm-report-caches --quiet'), $script); $this->assertSame( 1, substr_count($contents, 'business-mirror:sync-third-party-warehouse-storage --max-batches=0 --quiet'), @@ -103,6 +120,20 @@ class BusinessMirrorScheduleTest extends TestCase } } + public function test_shared_config_database_is_sized_for_reports_and_records_slow_queries(): void + { + $compose = file_get_contents(dirname(__DIR__, 3) . '/compose.jiangling.yml'); + $start = strpos($compose, " config_db:\n"); + $end = strpos($compose, " clickhouse:\n", $start); + $service = substr($compose, $start, $end - $start); + + $this->assertStringContainsString('--innodb-buffer-pool-size=4G', $service); + $this->assertStringContainsString('--slow-query-log=ON', $service); + $this->assertStringContainsString('--long-query-time=5', $service); + $this->assertStringContainsString('mem_limit: 6g', $service); + $this->assertStringContainsString('cpus: "4.00"', $service); + } + public function test_production_supplier_payable_maintenance_uses_the_business_mirror(): void { $compose = file_get_contents(dirname(__DIR__, 3) . '/compose.jiangling.yml'); diff --git a/tests/Unit/EcommerceFinance/StoreReceivingAlipayBillMirrorRepositoryTest.php b/tests/Unit/EcommerceFinance/StoreReceivingAlipayBillMirrorRepositoryTest.php index 05e262b2..f6f751e5 100644 --- a/tests/Unit/EcommerceFinance/StoreReceivingAlipayBillMirrorRepositoryTest.php +++ b/tests/Unit/EcommerceFinance/StoreReceivingAlipayBillMirrorRepositoryTest.php @@ -3,6 +3,7 @@ namespace Tests\Unit\EcommerceFinance; use Illuminate\Database\Schema\Blueprint; +use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Schema; use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository; @@ -333,7 +334,7 @@ class StoreReceivingAlipayBillMirrorRepositoryTest extends TestCase $this->assertSame(['income' => 100.25, 'expense' => 0.0, 'balance' => 100.25, 'count' => 1], $result['summary']); } - public function test_taobao_scoped_pages_recount_the_subset_and_observe_the_next_sync(): void + public function test_taobao_scoped_pages_share_one_aggregate_across_sync_state_churn(): void { $this->markReady(); $db = DB::connection('business_mirror'); @@ -361,9 +362,12 @@ class StoreReceivingAlipayBillMirrorRepositoryTest extends TestCase $db->table('ops_jst_original_bill_info')->insert($this->row(3, 7, '2026-08-03 09:00:00', '1', 4)); $state->update(['mirror_row_count' => 3, 'last_succeeded_at' => '2026-09-08 09:00:00']); - $this->assertSame(3, $this->repository->paginate([], 'occurred_at', 'desc', 1, 1)['total']); + $this->assertSame(2, $this->repository->paginate([], 'occurred_at', 'desc', 1, 1)['total']); $this->assertSame(1, $this->repository->paginate(['business_type' => '支出'], 'occurred_at', 'desc', 1, 1)['total']); + Cache::flush(); + $this->assertSame(3, $this->repository->paginate([], 'occurred_at', 'desc', 1, 1)['total']); + $state->update(['status' => 'running', 'mirror_row_count' => 2]); $this->assertSame(3, $this->repository->paginate([], 'occurred_at', 'desc', 1, 1)['total']); $state->update(['status' => 'failed']); @@ -372,7 +376,7 @@ class StoreReceivingAlipayBillMirrorRepositoryTest extends TestCase $db->flushQueryLog(); $db->enableQueryLog(); $this->assertSame(3, $this->repository->paginate([], 'occurred_at', 'desc', 1, 1)['total']); - $this->assertCount(1, array_filter($db->getQueryLog(), fn ($q) => str_contains(strtolower($q['query']), 'count(*)'))); + $this->assertCount(0, array_filter($db->getQueryLog(), fn ($q) => str_contains(strtolower($q['query']), 'count(*)'))); $db->disableQueryLog(); $state->update(['status' => 'succeeded', 'mirror_row_count' => null]); $this->assertSame(3, $this->repository->paginate([], 'occurred_at', 'desc', 1, 1)['total']);