公司店铺商标页面切换共享业务镜像 (#54)

app/Business/UpstreamBusinessInformation.php

@@ -4,6 +4,7 @@ namespace App\Business;
 
 use Illuminate\Pagination\LengthAwarePaginator;
 use InvalidArgumentException;
+use Modules\EcommerceFinance\Repositories\BusinessInformationMirrorRepository;
 use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory;
 use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
 
@@ -11,9 +12,10 @@ class UpstreamBusinessInformation
 {
     private const SOURCE = 'operations_replica';
 
-    public function __construct(private readonly UpstreamQueryCacheFactory $caches)
-    {
-    }
+    public function __construct(
+        private readonly UpstreamQueryCacheFactory $caches,
+        private readonly BusinessInformationMirrorRepository $mirror,
+    ) {}
 
     public function paginate(string $section, string $page, int $perPage = 50): LengthAwarePaginator
     {
@@ -64,7 +66,7 @@ class UpstreamBusinessInformation
 
     private function companyPage(int $page, int $perPage): array
     {
-        return $this->remember(
+        return $this->page(
             'company-information.read',
             $page,
             $perPage,
@@ -108,7 +110,7 @@ SQL);
 
     private function storePage(int $page, int $perPage): array
     {
-        return $this->remember(
+        return $this->page(
             'store-information.read',
             $page,
             $perPage,
@@ -156,7 +158,7 @@ SQL);
 
     private function trademarkPage(int $page, int $perPage): array
     {
-        return $this->remember(
+        return $this->page(
             'trademark-information.read',
             $page,
             $perPage,
@@ -191,8 +193,12 @@ SQL);
         ];
     }
 
-    private function remember(string $scope, int $page, int $perPage, callable $resolver): array
+    private function page(string $scope, int $page, int $perPage, callable $resolver): array
     {
+        if ($this->mirror->ready($scope)) {
+            return $this->normalize($scope, $this->mirror->paginate($scope, $page, $perPage));
+        }
+
         $cache = $this->caches->forSource(self::SOURCE);
 
         return $cache->remember($scope, compact('page', 'perPage'), function () use ($resolver) {
@@ -200,6 +206,22 @@ SQL);
         });
     }
 
+    /** @param array{data: array<int,array<string,mixed>>, total: int} $payload */
+    private function normalize(string $scope, array $payload): array
+    {
+        $mapper = match ($scope) {
+            'company-information.read' => fn (array $row): array => $this->mapCompany($row),
+            'store-information.read' => fn (array $row): array => $this->mapStore($row),
+            'trademark-information.read' => fn (array $row): array => $this->mapTrademark($row),
+            default => throw new InvalidArgumentException('未知的经营分析副本库资料类型：' . $scope),
+        };
+
+        return [
+            'data' => array_map($mapper, $payload['data']),
+            'total' => (int) $payload['total'],
+        ];
+    }
+
     private function mapCompany(array $row): array
     {
         $row = $this->toUtf8($row);

docs/上游业务镜像清单.md

@@ -12,7 +12,7 @@
 | --- | --- |
 | 已完成 | `operations_replica/rpt.Bulk_Price_Limit` 已进入类型化表 `product_price_mirror_rows`，商品价格页面已读取镜像并支持完整分页。 |
 | 本次建立 | `config/upstream_mirrors.php` 中登记全部当前代码引用的原表；通用表级同步框架负责建表、稳定游标、断点续传、增量重叠和全量校准。 |
-| 待逐域切换 | 其他业务 Repository 仍需从上游 SQL Server 查询改为读取对应镜像表。切换前保留旧快照作为回滚，不得宣称已经完成。 |
+| 已开始逐域切换 | 公司、店铺、商标页面在对应镜像完成首次全量后自动改读完整镜像表；首次全量前保留旧快照作为兼容回滚。其他业务域仍需逐项切换，不得宣称全项目已经完成。 |
 
 ## 经营分析副本库（operations_replica）
 

modules/EcommerceFinance/Repositories/BusinessInformationMirrorRepository.php

@@ -0,0 +1,150 @@
+<?php
+
+namespace Modules\EcommerceFinance\Repositories;
+
+use Illuminate\Database\ConnectionInterface;
+use InvalidArgumentException;
+use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
+use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
+use RuntimeException;
+
+class BusinessInformationMirrorRepository
+{
+    private const DATASETS = [
+        'company-information.read' => 'operations_replica.rpt.fys_companyInfo',
+        'store-information.read' => 'operations_replica.rpt.fys_shopInfo',
+        'trademark-information.read' => 'operations_replica.rpt.fys_TrademarkInfo',
+    ];
+
+    private readonly ConnectionInterface $connection;
+
+    public function __construct(
+        private readonly BusinessMirrorDatasetRegistry $datasets,
+        private readonly GenericBusinessMirrorRepository $mirror,
+    ) {
+        $this->connection = $mirror->connection();
+    }
+
+    public function ready(string $scope): bool
+    {
+        return $this->mirror->ready($this->dataset($scope));
+    }
+
+    /** @return array{data: array<int,array<string,mixed>>, total: int} */
+    public function paginate(string $scope, int $page, int $perPage): array
+    {
+        if (! $this->ready($scope)) {
+            throw new RuntimeException('business_mirror_dataset_not_ready');
+        }
+
+        $page = max($page, 1);
+        $perPage = min(max($perPage, 1), 100);
+
+        return match ($scope) {
+            'company-information.read' => $this->companies($page, $perPage),
+            'store-information.read' => $this->stores($page, $perPage),
+            'trademark-information.read' => $this->trademarks($page, $perPage),
+            default => throw new InvalidArgumentException('未知的业务资料镜像范围：' . $scope),
+        };
+    }
+
+    /** @return array{data: array<int,array<string,mixed>>, total: int} */
+    private function companies(int $page, int $perPage): array
+    {
+        $query = $this->connection->table('ops_fys_company_info')
+            ->selectRaw(<<<'SQL'
+Id AS id,
+unifiedSocialCreditCode AS unified_social_credit_code,
+businessLicenseName AS company_name,
+registeredCapital AS subscribed_capital,
+businessLicenseType AS company_type,
+IncorporationDate AS establishment_date,
+legalRepresentative AS legal_representative,
+registeredAddress AS registered_address,
+businessScope AS business_scope,
+accountSerialNumber AS basic_deposit_account_number,
+companyStatus AS company_status,
+financialManager AS financial_officer,
+supervisor AS supervisor,
+taxCategoryAndType AS taxpayer_category,
+NULL AS shareholders
+SQL);
+
+        return $this->page($query, $page, $perPage, 'Id');
+    }
+
+    /** @return array{data: array<int,array<string,mixed>>, total: int} */
+    private function stores(int $page, int $perPage): array
+    {
+        $latestCompanies = $this->connection->table('ops_fys_company_info')
+            ->selectRaw('unifiedSocialCreditCode, MAX(Id) AS latest_id')
+            ->whereNotNull('unifiedSocialCreditCode')
+            ->where('unifiedSocialCreditCode', '<>', '')
+            ->groupBy('unifiedSocialCreditCode');
+
+        $query = $this->connection->table('ops_fys_shop_info AS s')
+            ->leftJoinSub($latestCompanies, 'company_keys', function ($join): void {
+                $join->on('company_keys.unifiedSocialCreditCode', '=', 's.unifiedSocialCreditCode');
+            })
+            ->leftJoin('ops_fys_company_info AS c', 'c.Id', '=', 'company_keys.latest_id')
+            ->selectRaw(<<<'SQL'
+s.Id AS id,
+s.openTime AS opened_at,
+s.erpShopName AS erp_store_name,
+s.unifiedSocialCreditCode AS unified_social_credit_code,
+c.businessLicenseName AS company_name,
+s.trademarkNumber AS trademark_number,
+NULL AS trademark_name,
+s.platform AS platform,
+s.responsibleDepartment AS profit_center,
+s.shopHead AS store_manager,
+s.shopState AS store_status,
+s.shopType AS store_type,
+s.channelType AS channel_mode
+SQL);
+
+        return $this->page($query, $page, $perPage, 's.Id');
+    }
+
+    /** @return array{data: array<int,array<string,mixed>>, total: int} */
+    private function trademarks(int $page, int $perPage): array
+    {
+        $query = $this->connection->table('ops_fys_trademark_info')
+            ->where('isDelete', 0)
+            ->selectRaw(<<<'SQL'
+Id AS id,
+trademarkName AS trademark_name,
+internationalClassification AS trademark_type,
+trademarkStatus AS trademark_status,
+registrationDate AS registration_date,
+termOfValidity AS expiry_date,
+remark AS notes
+SQL);
+
+        return $this->page($query, $page, $perPage, 'Id');
+    }
+
+    /** @return array{data: array<int,array<string,mixed>>, total: int} */
+    private function page($query, int $page, int $perPage, string $orderColumn): array
+    {
+        $total = (clone $query)->count();
+        $rows = $query->orderByDesc($orderColumn)
+            ->forPage($page, $perPage)
+            ->get()
+            ->map(static fn (object $row): array => (array) $row)
+            ->all();
+
+        return ['data' => $rows, 'total' => $total];
+    }
+
+    private function dataset(string $scope): BusinessMirrorDataset
+    {
+        $key = self::DATASETS[$scope] ?? null;
+
+        if ($key === null) {
+            throw new InvalidArgumentException('未知的业务资料镜像范围：' . $scope);
+        }
+
+        return $this->datasets->get($key);
+    }
+}

tests/Unit/EcommerceFinance/BusinessInformationMirrorRepositoryTest.php

@@ -0,0 +1,125 @@
+<?php
+
+namespace Tests\Unit\EcommerceFinance;
+
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+use Modules\EcommerceFinance\Repositories\BusinessInformationMirrorRepository;
+use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
+use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
+use Tests\TestCase;
+
+class BusinessInformationMirrorRepositoryTest extends TestCase
+{
+    private BusinessInformationMirrorRepository $repository;
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+
+        config([
+            'database.connections.business_mirror' => [
+                'driver' => 'sqlite',
+                'database' => ':memory:',
+                'prefix' => '',
+            ],
+            'upstream_mirrors.connection' => 'business_mirror',
+            'upstream_mirrors.datasets' => [
+                'operations_replica.rpt.fys_companyInfo' => $this->definition('fys_companyInfo', 'ops_fys_company_info'),
+                'operations_replica.rpt.fys_shopInfo' => $this->definition('fys_shopInfo', 'ops_fys_shop_info'),
+                'operations_replica.rpt.fys_TrademarkInfo' => $this->definition('fys_TrademarkInfo', 'ops_fys_trademark_info'),
+            ],
+        ]);
+        DB::purge('business_mirror');
+
+        $this->createTables();
+        $mirror = new GenericBusinessMirrorRepository(app('db'));
+        $this->repository = new BusinessInformationMirrorRepository(
+            new BusinessMirrorDatasetRegistry(),
+            $mirror,
+        );
+    }
+
+    public function test_it_reads_all_three_ready_mirrors_with_pagination_and_business_filters(): void
+    {
+        $connection = DB::connection('business_mirror');
+        $connection->table('ops_fys_company_info')->insert([
+            $this->company(1, '统一代码', '旧公司'),
+            $this->company(2, '统一代码', '新公司'),
+            $this->company(3, '其他代码', '其他公司'),
+        ]);
+        $connection->table('ops_fys_shop_info')->insert([
+            ['Id' => 8, 'openTime' => '2026-08-14 10:00:00', 'erpShopName' => '测试店', 'unifiedSocialCreditCode' => '统一代码', 'trademarkNumber' => 'TM', 'platform' => '天猫', 'responsibleDepartment' => '电商', 'shopHead' => '负责人', 'shopState' => '0', 'shopType' => '直营', 'channelType' => '线上'],
+        ]);
+        $connection->table('ops_fys_trademark_info')->insert([
+            ['Id' => 5, 'trademarkName' => '有效商标', 'internationalClassification' => '20', 'trademarkStatus' => '1', 'registrationDate' => '2020-01-01', 'termOfValidity' => '2030-01-01', 'remark' => null, 'isDelete' => 0],
+            ['Id' => 6, 'trademarkName' => '已删除', 'internationalClassification' => '20', 'trademarkStatus' => '0', 'registrationDate' => null, 'termOfValidity' => null, 'remark' => null, 'isDelete' => 1],
+        ]);
+        $this->markReady();
+
+        $companies = $this->repository->paginate('company-information.read', 2, 2);
+        $store = $this->repository->paginate('store-information.read', 1, 20);
+        $trademarks = $this->repository->paginate('trademark-information.read', 1, 20);
+
+        $this->assertSame(3, $companies['total']);
+        $this->assertSame(1, $companies['data'][0]['id']);
+        $this->assertSame('新公司', $store['data'][0]['company_name']);
+        $this->assertSame(1, $trademarks['total']);
+        $this->assertSame('有效商标', $trademarks['data'][0]['trademark_name']);
+    }
+
+    public function test_dataset_is_not_ready_until_initial_full_sync_is_marked_complete(): void
+    {
+        $this->assertFalse($this->repository->ready('company-information.read'));
+        $this->markReady(['operations_replica.rpt.fys_companyInfo']);
+        $this->assertTrue($this->repository->ready('company-information.read'));
+    }
+
+    private function createTables(): void
+    {
+        $schema = Schema::connection('business_mirror');
+        $schema->create('business_mirror_table_states', function (Blueprint $table): void {
+            $table->string('dataset_key')->primary();
+            $table->dateTime('initial_completed_at')->nullable();
+        });
+        $schema->create('ops_fys_company_info', function (Blueprint $table): void {
+            $table->integer('Id'); $table->string('unifiedSocialCreditCode')->nullable(); $table->string('businessLicenseName')->nullable();
+            $table->string('registeredCapital')->nullable(); $table->string('businessLicenseType')->nullable(); $table->dateTime('IncorporationDate')->nullable();
+            $table->string('legalRepresentative')->nullable(); $table->string('registeredAddress')->nullable(); $table->text('businessScope')->nullable();
+            $table->string('accountSerialNumber')->nullable(); $table->string('companyStatus')->nullable(); $table->string('financialManager')->nullable();
+            $table->string('supervisor')->nullable(); $table->string('taxCategoryAndType')->nullable();
+        });
+        $schema->create('ops_fys_shop_info', function (Blueprint $table): void {
+            $table->integer('Id'); $table->dateTime('openTime')->nullable(); $table->string('erpShopName')->nullable(); $table->string('unifiedSocialCreditCode')->nullable();
+            $table->string('trademarkNumber')->nullable(); $table->string('platform')->nullable(); $table->string('responsibleDepartment')->nullable();
+            $table->string('shopHead')->nullable(); $table->string('shopState')->nullable(); $table->string('shopType')->nullable(); $table->string('channelType')->nullable();
+        });
+        $schema->create('ops_fys_trademark_info', function (Blueprint $table): void {
+            $table->integer('Id'); $table->string('trademarkName')->nullable(); $table->string('internationalClassification')->nullable();
+            $table->string('trademarkStatus')->nullable(); $table->dateTime('registrationDate')->nullable(); $table->dateTime('termOfValidity')->nullable();
+            $table->text('remark')->nullable(); $table->integer('isDelete')->default(0);
+        });
+    }
+
+    private function markReady(?array $keys = null): void
+    {
+        $keys ??= array_keys((array) config('upstream_mirrors.datasets'));
+        foreach ($keys as $key) {
+            DB::connection('business_mirror')->table('business_mirror_table_states')->insert([
+                'dataset_key' => $key,
+                'initial_completed_at' => '2026-08-14 12:00:00',
+            ]);
+        }
+    }
+
+    private function definition(string $object, string $target): array
+    {
+        return ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => $object, 'target' => $target, 'keys' => ['Id']];
+    }
+
+    private function company(int $id, string $code, string $name): array
+    {
+        return ['Id' => $id, 'unifiedSocialCreditCode' => $code, 'businessLicenseName' => $name, 'registeredCapital' => null, 'businessLicenseType' => null, 'IncorporationDate' => null, 'legalRepresentative' => null, 'registeredAddress' => null, 'businessScope' => null, 'accountSerialNumber' => null, 'companyStatus' => '0', 'financialManager' => null, 'supervisor' => null, 'taxCategoryAndType' => null];
+    }
+}

tests/Unit/EcommerceFinance/UpstreamBusinessInformationTest.php

@@ -4,6 +4,8 @@ namespace Tests\Unit\EcommerceFinance;
 
 use App\Business\UpstreamBusinessInformation;
 use Closure;
+use Mockery;
+use Modules\EcommerceFinance\Repositories\BusinessInformationMirrorRepository;
 use Modules\EcommerceFinance\Services\UpstreamQueryCache;
 use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory;
 use Tests\TestCase;
@@ -44,7 +46,10 @@ class UpstreamBusinessInformationTest extends TestCase
             }
         };
 
-        $information = new class($caches) extends UpstreamBusinessInformation {
+        $mirror = Mockery::mock(BusinessInformationMirrorRepository::class);
+        $mirror->shouldReceive('ready')->once()->with($scope)->andReturnFalse();
+
+        $information = new class($caches, $mirror) extends UpstreamBusinessInformation {
             public array $snapshots = [];
 
             public function snapshot(string $scope, array $parameters): array
@@ -65,6 +70,43 @@ class UpstreamBusinessInformationTest extends TestCase
         $this->assertSame($expected, $information->snapshots[0]);
     }
 
+    /**
+     * @dataProvider businessInformationPages
+     */
+    public function test_ready_business_mirror_is_used_without_registering_a_page_snapshot(
+        string $section,
+        string $page,
+        string $scope,
+    ): void {
+        $cache = new class extends UpstreamQueryCache {
+            public function __construct() {}
+
+            public function remember(string $scope, array $parameters, Closure $resolver): mixed
+            {
+                throw new \RuntimeException('镜像就绪后不应再读取分页快照。');
+            }
+        };
+        $caches = new class($cache) extends UpstreamQueryCacheFactory {
+            public function __construct(private readonly UpstreamQueryCache $cache) {}
+
+            public function forSource(string $source): UpstreamQueryCache { return $this->cache; }
+        };
+        $mirror = Mockery::mock(BusinessInformationMirrorRepository::class);
+        $mirror->shouldReceive('ready')->once()->with($scope)->andReturnTrue();
+        $mirror->shouldReceive('paginate')->once()->with($scope, 2, 25)->andReturn([
+            'data' => [['id' => 10]],
+            'total' => 51,
+        ]);
+
+        $this->app['request']->merge(['page' => 2]);
+        $paginator = (new UpstreamBusinessInformation($caches, $mirror))
+            ->paginate($section, $page, 25);
+
+        $this->assertSame(2, $paginator->currentPage());
+        $this->assertSame(51, $paginator->total());
+        $this->assertSame(10, $paginator->items()[0]->id);
+    }
+
     public static function businessInformationPages(): array
     {
         return [