WuYongJie/CaiWuXiTong
1
0
Fork 0
Code Issues Pull Requests 5 Packages Projects Releases Wiki Activity

fix: refresh operations replica snapshots #7

Merged
WuYongJie merged 1 commits from hotfix/WuYongJie-operations-replica-snapshots into main 2026-08-04 15:41:50 +08:00
Conversation 0 Commits 1 Files Changed 6 +231 -29
6 changed files with 231 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

95
app/Business/UpstreamBusinessInformation.php
View File

@ -3,6 +3,7 @@
namespace App\Business;
use Illuminate\Pagination\LengthAwarePaginator;
use InvalidArgumentException;
use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
@ -39,11 +40,42 @@ class UpstreamBusinessInformation
);
}
/**
* 将已登记的经营分析副本库查询还原成不经过页面缓存的固定只读查询。
*
* 生产后台刷新器使用这个入口生成共享快照；页面仍然只能通过
* UpstreamQueryCache 读取快照，不能绕过缓存直接访问上游。
*
* @return array{data: array<int,array<string,mixed>>, total: int}
*/
public function snapshot(string $scope, array $parameters): array
{
$page = max((int) ($parameters['page'] ?? 1), 1);
$perPage = min(max((int) ($parameters['perPage'] ?? 50), 1), 100);
$query = new UpstreamReadOnlyQuery(self::SOURCE);
return match ($scope) {
'company-information.read' => $this->companySnapshot($query, $page, $perPage),
'store-information.read' => $this->storeSnapshot($query, $page, $perPage),
'trademark-information.read' => $this->trademarkSnapshot($query, $page, $perPage),
default => throw new InvalidArgumentException('未知的经营分析副本库快照类型：' . $scope),
};
}
private function companyPage(int $page, int $perPage): array
{
return $this->remember('company-information.read', $page, $perPage, function (UpstreamReadOnlyQuery $query) use ($page, $perPage) {
$offset = ($page - 1) * $perPage;
$rows = $query->select(<<<SQL
return $this->remember(
'company-information.read',
$page,
$perPage,
fn () => $this->snapshot('company-information.read', compact('page', 'perPage')),
);
}
private function companySnapshot(UpstreamReadOnlyQuery $query, int $page, int $perPage): array
{
$offset = ($page - 1) * $perPage;
$rows = $query->select(<<<SQL
SELECT
Id AS id,
unifiedSocialCreditCode AS unified_social_credit_code,
@ -65,18 +97,26 @@ ORDER BY Id DESC
OFFSET {$offset} ROWS FETCH NEXT {$perPage} ROWS ONLY
SQL);
return [
'data' => array_map(fn (array $row) => $this->mapCompany($row), $rows),
'total' => (int) $query->scalar('SELECT COUNT(*) FROM rpt.fys_companyInfo'),
];
});
return [
'data' => array_map(fn (array $row) => $this->mapCompany($row), $rows),
'total' => (int) $query->scalar('SELECT COUNT(*) FROM rpt.fys_companyInfo'),
];
}
private function storePage(int $page, int $perPage): array
{
return $this->remember('store-information.read', $page, $perPage, function (UpstreamReadOnlyQuery $query) use ($page, $perPage) {
$offset = ($page - 1) * $perPage;
$rows = $query->select(<<<SQL
return $this->remember(
'store-information.read',
$page,
$perPage,
fn () => $this->snapshot('store-information.read', compact('page', 'perPage')),
);
}
private function storeSnapshot(UpstreamReadOnlyQuery $query, int $page, int $perPage): array
{
$offset = ($page - 1) * $perPage;
$rows = $query->select(<<<SQL
SELECT
s.Id AS id,
s.openTime AS opened_at,
@ -102,18 +142,26 @@ ORDER BY s.Id DESC
OFFSET {$offset} ROWS FETCH NEXT {$perPage} ROWS ONLY
SQL);
return [
'data' => array_map(fn (array $row) => $this->mapStore($row), $rows),
'total' => (int) $query->scalar('SELECT COUNT(*) FROM rpt.fys_shopInfo'),
];
});
return [
'data' => array_map(fn (array $row) => $this->mapStore($row), $rows),
'total' => (int) $query->scalar('SELECT COUNT(*) FROM rpt.fys_shopInfo'),
];
}
private function trademarkPage(int $page, int $perPage): array
{
return $this->remember('trademark-information.read', $page, $perPage, function (UpstreamReadOnlyQuery $query) use ($page, $perPage) {
$offset = ($page - 1) * $perPage;
$rows = $query->select(<<<SQL
return $this->remember(
'trademark-information.read',
$page,
$perPage,
fn () => $this->snapshot('trademark-information.read', compact('page', 'perPage')),
);
}
private function trademarkSnapshot(UpstreamReadOnlyQuery $query, int $page, int $perPage): array
{
$offset = ($page - 1) * $perPage;
$rows = $query->select(<<<SQL
SELECT
Id AS id,
trademarkName AS trademark_name,
@ -128,11 +176,10 @@ ORDER BY Id DESC
OFFSET {$offset} ROWS FETCH NEXT {$perPage} ROWS ONLY
SQL);
return [
'data' => array_map(fn (array $row) => $this->mapTrademark($row), $rows),
'total' => (int) $query->scalar('SELECT COUNT(*) FROM rpt.fys_TrademarkInfo WHERE isDelete = 0'),
];
});
return [
'data' => array_map(fn (array $row) => $this->mapTrademark($row), $rows),
'total' => (int) $query->scalar('SELECT COUNT(*) FROM rpt.fys_TrademarkInfo WHERE isDelete = 0'),
];
}
private function remember(string $scope, int $page, int $perPage, callable $resolver): array

23
app/Http/Controllers/Business/Pages.php
View File

@ -145,6 +145,7 @@ class Pages extends Controller
$rows = collect();
$tabValue = null;
$tabColumn = null;
$upstreamError = null;
if ($page && $content === null) {
$descriptor = TableRegistry::describe(
@ -155,11 +156,21 @@ class Pages extends Controller
if ($descriptor) {
if ($descriptor->readOnly) {
$rows = app(UpstreamBusinessInformation::class)->paginate(
$section['slug'],
$page['slug'],
50,
);
try {
$rows = app(UpstreamBusinessInformation::class)->paginate(
$section['slug'],
$page['slug'],
50,
);
} catch (ServiceUnavailableHttpException $exception) {
// 首份快照尚未生成或某个来源暂时不可用时，保留页面、
// 交代原因并允许后台刷新，而不是把整个业务页变成通用 503。
$rows = new LengthAwarePaginator([], 0, 50, 1, [
'path' => request()->url(),
'query' => request()->query(),
]);
$upstreamError = $exception->getMessage();
}
return [
'section' => $section,
@ -169,6 +180,7 @@ class Pages extends Controller
'content' => $content,
'descriptor' => $descriptor,
'rows' => $rows,
'upstreamError' => $upstreamError,
'tabColumn' => null,
'tabValue' => null,
'edgeBanding' => $this->edgeBandingContent($page['slug']),
@ -205,6 +217,7 @@ class Pages extends Controller
'content' => $content,
'descriptor' => $descriptor,
'rows' => $rows,
'upstreamError' => $upstreamError,
'tabColumn' => $tabColumn,
'tabValue' => $tabValue,
'edgeBanding' => $page ? $this->edgeBandingContent($page['slug']) : null,

2
modules/EcommerceFinance/Providers/Main.php
View File

@ -4,6 +4,7 @@ namespace Modules\EcommerceFinance\Providers;
use Illuminate\Support\ServiceProvider as Provider;
use Modules\EcommerceFinance\Services\JianSunSnapshotRefresher;
use Modules\EcommerceFinance\Services\OperationsReplicaSnapshotRefresher;
use Modules\EcommerceFinance\Services\UpstreamSnapshotRefresherRegistry;
use Modules\EcommerceFinance\Services\UpstreamSourceRegistry;
@ -22,6 +23,7 @@ class Main extends Provider
$app->make(UpstreamSourceRegistry::class)
);
$registry->register('jiansun', fn () => $app->make(JianSunSnapshotRefresher::class));
$registry->register('operations_replica', fn () => $app->make(OperationsReplicaSnapshotRefresher::class));
return $registry;
});

37
modules/EcommerceFinance/Services/OperationsReplicaSnapshotRefresher.php Normal file
View File

@ -0,0 +1,37 @@
<?php
namespace Modules\EcommerceFinance\Services;
use App\Business\UpstreamBusinessInformation;
use Modules\EcommerceFinance\Contracts\UpstreamSnapshotRefresher;
/**
* 将经营分析副本库已登记的业务信息查询还原为共享快照刷新任务。
*/
class OperationsReplicaSnapshotRefresher implements UpstreamSnapshotRefresher
{
private const SOURCE = 'operations_replica';
public function __construct(
private readonly UpstreamQueryCacheFactory $caches,
private readonly UpstreamBusinessInformation $information,
) {
}
public function sourceKey(): string
{
return self::SOURCE;
}
public function refresh(array $entry): string
{
$scope = (string) ($entry['scope'] ?? '');
$parameters = (array) ($entry['parameters'] ?? []);
return $this->caches->forSource(self::SOURCE)->refresh(
$scope,
$parameters,
fn () => $this->information->snapshot($scope, $parameters),
);
}
}

5
resources/views/business/pages/show.blade.php
View File

@ -401,6 +401,11 @@
@if ($errors->has('import'))
<div class="biz-alert biz-alert-error" role="alert">{{ $errors->first('import') }}</div>
@endif
@if (!empty($upstreamError))
<div class="biz-alert biz-alert-error" role="alert">{{ $upstreamError }}</div>
@endif
@if ($tabs->isNotEmpty())
<nav class="business-tab-bar" aria-label="{{ $page['title'] }}">
@foreach ($tabs as $tab)

98
tests/Unit/EcommerceFinance/OperationsReplicaSnapshotRefresherTest.php Normal file
View File

@ -0,0 +1,98 @@
<?php
namespace Tests\Unit\EcommerceFinance;
use App\Business\UpstreamBusinessInformation;
use Closure;
use InvalidArgumentException;
use Modules\EcommerceFinance\Services\OperationsReplicaSnapshotRefresher;
use Modules\EcommerceFinance\Services\UpstreamQueryCache;
use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory;
use PHPUnit\Framework\TestCase;
use ReflectionClass;
class OperationsReplicaSnapshotRefresherTest extends TestCase
{
/**
* @dataProvider snapshotScopes
*/
public function test_background_refresh_uses_the_same_fixed_business_snapshot_query(
string $scope,
array $parameters,
): void {
$cache = new class extends UpstreamQueryCache {
public array $calls = [];
public mixed $value = null;
public function __construct()
{
}
public function refresh(string $scope, array $parameters, Closure $resolver): string
{
$this->calls[] = compact('scope', 'parameters');
$this->value = $resolver();
return 'fresh';
}
};
$caches = new class($cache) extends UpstreamQueryCacheFactory {
public string $requestedSource = '';
public function __construct(private readonly UpstreamQueryCache $cache)
{
}
public function forSource(string $source): UpstreamQueryCache
{
$this->requestedSource = $source;
return $this->cache;
}
};
$information = new class extends UpstreamBusinessInformation {
public array $calls = [];
public function __construct()
{
}
public function snapshot(string $scope, array $parameters): array
{
$this->calls[] = compact('scope', 'parameters');
return ['scope' => $scope, 'parameters' => $parameters];
}
};
$refresher = new OperationsReplicaSnapshotRefresher($caches, $information);
$this->assertSame('operations_replica', $refresher->sourceKey());
$this->assertSame('fresh', $refresher->refresh(compact('scope', 'parameters')));
$this->assertSame('operations_replica', $caches->requestedSource);
$this->assertSame([compact('scope', 'parameters')], $cache->calls);
$this->assertSame([compact('scope', 'parameters')], $information->calls);
$this->assertSame(['scope' => $scope, 'parameters' => $parameters], $cache->value);
}
public static function snapshotScopes(): array
{
return [
'company' => ['company-information.read', ['page' => 2, 'perPage' => 50]],
'store' => ['store-information.read', ['page' => 1, 'perPage' => 25]],
'trademark' => ['trademark-information.read', ['page' => 3, 'perPage' => 100]],
];
}
public function test_unknown_business_snapshot_scope_is_rejected_before_connecting_to_an_upstream(): void
{
/** @var UpstreamBusinessInformation $information */
$information = (new ReflectionClass(UpstreamBusinessInformation::class))->newInstanceWithoutConstructor();
$this->expectException(InvalidArgumentException::class);
$information->snapshot('unknown.scope', []);
}
}