WuYongJie
e144a3562c
## 任务占用与边界 - 任务名称：修复上游业务页面快照完整性 - 实际开发者：WuYongJie - 任务来源/团队分工：项目负责人反馈生产业务页面空数据、503 与首份快照长期未生成 - 目标结果：所有已接入上游的页面能够公平生成列表、汇总和筛选项快照；未放行范围明确列出并走生产授权 - 预计修改路径：modules/EcommerceFinance/Services/UpstreamQueryCache.php；tests/Unit/EcommerceFinance/JianSunQueryCacheTest.php - 预计页面或路由：所有使用 upstream snapshots 的业务页面与 workspace-data 接口 - 预计 Controller / Repository：不修改业务 Repository；修复共享 UpstreamQueryCache 调度 - 数据源、表/视图和字段：jiansun 已有固定只读 Repository；不新增表、不改字段 - 上游 source_key / scope：jiansun；goods.*、invoice-deductions.*、taobao-alipay-bills.* 及既有快照范围 - 权限名、菜单键或共享组件：不修改 - 明确不修改的范围：不写生产数据库、不迁移、不部署、不修改上游 SQL 与业务金额逻辑 ## 开工碰撞检查 - [x] Codex 已执行 scripts/codex-task-preflight.ps1 - [x] 已检查最新 main 和所有远程任务分支 - [x] 已搜索主线既有页面、接口、Repository 和数据查询 - [x] 未发现当前计划文件重叠 相关分支、重叠项与协调结论：两个历史本地共享数据库分支涉及 Pages.php 等文件；本任务缩小为 UpstreamQueryCache 与对应单测，不与其重叠。 ## 实际改动 - 完成内容：WIP - Codex 实际修改的文件：WIP - 新增或变更的公共契约：WIP - 与原计划的差异：WIP ## 验证证据 - 正常流程：WIP - 空数据/错误输入/无权限场景：WIP - 受影响旧功能：WIP - 自动测试及结果：WIP - 实际页面、接口或只读业务查询结果：WIP ## 风险检查 - [ ] 不涉及金额、权限、审批、数据库迁移、生产配置或数据修复 - [x] 如涉及上述高风险内容，已在下方写明影响、回滚和所需授权 - [x] 未提交 .env、密码、Token、私钥、真实财务数据、生产日志或构建产物 - [x] 前端改动符合 DESIGN.md；无前端改动，不适用 高风险影响、回滚和授权状态：代码修复本身不写生产数据；生产 allowed_scopes 调整、快照生成和部署必须另行取得明确授权。 ## 合并前复查 - [ ] 已重新拉取远程引用并用实际修改范围再次运行碰撞检查 - [ ] PR 不再是 WIP:，所有测试结果已经补齐 - [ ] Codex 已重新读取原始需求和验收标准，并逐文件审核完整 diff - [ ] Codex 自审确认无越界修改、重复实现、敏感内容、未处理冲突和未验证结果 - [ ] 合并顺序不会覆盖其他未合并任务 - [ ] 已授权当前任务的 Codex执行 Squash merge、验证 main 并清理任务分支 - [ ] 如涉及生产迁移、数据修复、删除、覆盖配置或部署，已另行取得对应高风险授权 Reviewed-on: http://git.1080p.hk/WuYongJie/CaiWuXiTong/pulls/2
673 lines
22 KiB
PHP
673 lines
22 KiB
PHP
<?php
|
||
|
||
namespace Modules\EcommerceFinance\Services;
|
||
|
||
use Closure;
|
||
use Illuminate\Cache\CacheManager;
|
||
use Illuminate\Contracts\Cache\Repository;
|
||
use Illuminate\Contracts\Cache\Lock;
|
||
use Illuminate\Contracts\Cache\LockTimeoutException;
|
||
use Illuminate\Support\Facades\Log;
|
||
use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 任意上游数据源共用的共享快照、请求合并、限流和熔断实现。
|
||
*
|
||
* 所有缓存键和压力预算都包含 source，确保一个上游故障或繁忙时不会
|
||
* 阻塞其他上游。页面命中过期快照时立即返回旧值，由后台任务刷新。
|
||
*/
|
||
class UpstreamQueryCache
|
||
{
|
||
private const CACHE_VERSION = 'v2';
|
||
|
||
private readonly string $source;
|
||
|
||
private readonly string $label;
|
||
|
||
private readonly UpstreamSourceRegistry $sources;
|
||
|
||
private readonly Repository $cache;
|
||
|
||
private ?array $lastReadState = null;
|
||
|
||
private ?array $accessContext = null;
|
||
|
||
public function __construct(
|
||
UpstreamSourceRegistry $sources,
|
||
string $source,
|
||
CacheManager $cacheManager,
|
||
) {
|
||
$definition = $sources->definition($source);
|
||
$this->sources = $sources;
|
||
$this->source = $definition['key'];
|
||
$this->label = $definition['label'];
|
||
$this->cache = $cacheManager->store(
|
||
(string) config('upstreams.snapshots.cache_store', 'upstream_shared')
|
||
);
|
||
}
|
||
|
||
public function sourceKey(): string
|
||
{
|
||
return $this->source;
|
||
}
|
||
|
||
public function setting(string $path, mixed $default = null): mixed
|
||
{
|
||
return $this->sources->value($this->source, $path, $default);
|
||
}
|
||
|
||
/**
|
||
* 读取共享快照；过期快照采用 stale-while-revalidate，不阻塞页面。
|
||
*/
|
||
public function remember(string $scope, array $parameters, Closure $resolver): mixed
|
||
{
|
||
if (! (bool) $this->setting('enabled', false)) {
|
||
throw new ServiceUnavailableHttpException(60, '该上游数据同步尚未启用，请联系管理员。');
|
||
}
|
||
|
||
if ($this->usesDirectDevelopmentRead()) {
|
||
$value = $resolver();
|
||
$this->setReadState('direct', [
|
||
'cached_at' => now()->toIso8601String(),
|
||
]);
|
||
|
||
return $value;
|
||
}
|
||
|
||
if (! $this->isScopeAllowed($scope)) {
|
||
throw new ServiceUnavailableHttpException(30, '该数据范围仍在低压力验证中，请稍后再试。');
|
||
}
|
||
|
||
$parameters = $this->prepareParameters($parameters);
|
||
$key = $this->cacheKey($scope, $parameters);
|
||
$cached = $this->cache->get($key);
|
||
|
||
if ($this->hasValue($cached)) {
|
||
$this->register($scope, $parameters, false);
|
||
$this->setReadState('fresh', $cached);
|
||
|
||
return $cached['value'];
|
||
}
|
||
|
||
$stale = $this->cache->get($key . ':stale');
|
||
|
||
if ($this->hasValue($stale)) {
|
||
$this->register($scope, $parameters, true);
|
||
$this->setReadState('stale', $stale);
|
||
|
||
return $stale['value'];
|
||
}
|
||
|
||
$this->register($scope, $parameters, true);
|
||
|
||
if (! (bool) config('upstreams.snapshots.allow_inline_refresh', false)) {
|
||
throw new ServiceUnavailableHttpException(
|
||
15,
|
||
'数据同步请求已登记，后台正在准备首份共享数据，请稍后刷新。'
|
||
);
|
||
}
|
||
|
||
$resolved = $this->resolve($scope, $parameters, $resolver, false);
|
||
$payload = $this->cache->get($key) ?: $this->cache->get($key . ':stale');
|
||
$this->setReadState($resolved['status'], is_array($payload) ? $payload : []);
|
||
|
||
return $resolved['value'];
|
||
}
|
||
|
||
/**
|
||
* 供 API 响应暴露数据新鲜度，不改变既有 JSON 结构。
|
||
*/
|
||
public function responseHeaders(): array
|
||
{
|
||
if ($this->lastReadState === null) {
|
||
return [];
|
||
}
|
||
|
||
return array_filter([
|
||
'X-Caiwu-Data-Status' => $this->lastReadState['status'] ?? null,
|
||
'X-Caiwu-Data-Synced-At' => $this->lastReadState['cached_at'] ?? null,
|
||
]);
|
||
}
|
||
|
||
/**
|
||
* @return string fresh|stale
|
||
*/
|
||
public function refresh(string $scope, array $parameters, Closure $resolver): string
|
||
{
|
||
$parameters = $this->prepareParameters($parameters);
|
||
|
||
if (! (bool) $this->setting('enabled', false) || ! $this->isScopeAllowed($scope)) {
|
||
return 'stale';
|
||
}
|
||
|
||
return $this->resolve($scope, $parameters, $resolver, true)['status'];
|
||
}
|
||
|
||
/**
|
||
* 返回本来源最近实际访问且已经到期的查询。
|
||
*/
|
||
public function activeEntries(?int $limit = null): array
|
||
{
|
||
if (! (bool) $this->setting('enabled', false)) {
|
||
return [];
|
||
}
|
||
|
||
$now = time();
|
||
$activeTtl = max((int) $this->setting('snapshots.active_ttl_seconds', 21600), 900);
|
||
$entries = $this->cache->get($this->registryKey(), []);
|
||
|
||
if (! is_array($entries)) {
|
||
return [];
|
||
}
|
||
|
||
$entries = array_filter($entries, function (mixed $entry) use ($now, $activeTtl): bool {
|
||
return is_array($entry)
|
||
&& isset($entry['seen_at'], $entry['scope'], $entry['parameters'])
|
||
&& $this->isScopeAllowed((string) $entry['scope'])
|
||
&& ($now - (int) $entry['seen_at']) <= $activeTtl
|
||
&& $this->isDue($entry, $now);
|
||
});
|
||
|
||
usort($entries, function (array $a, array $b): int {
|
||
$urgent = ((int) ($b['urgent'] ?? 0)) <=> ((int) ($a['urgent'] ?? 0));
|
||
|
||
if ($urgent !== 0) {
|
||
return $urgent;
|
||
}
|
||
|
||
// 首份快照和过期快照都标记为 urgent。对于这些请求，必须让
|
||
// 等待最久的查询先执行；否则持续出现的新列表查询会凭借下方
|
||
// 的类型优先级，长期挤掉同一页面的 summary/options，页面就会
|
||
// 一直停在“列表有了，但汇总或筛选项仍 503”的半完成状态。
|
||
if ((bool) ($a['urgent'] ?? false)) {
|
||
$oldest = ((int) ($a['seen_at'] ?? 0)) <=> ((int) ($b['seen_at'] ?? 0));
|
||
|
||
if ($oldest !== 0) {
|
||
return $oldest;
|
||
}
|
||
}
|
||
|
||
$priority = $this->priorityFor((string) $b['scope']) <=> $this->priorityFor((string) $a['scope']);
|
||
|
||
if ($priority !== 0) {
|
||
return $priority;
|
||
}
|
||
|
||
return ((int) $b['seen_at']) <=> ((int) $a['seen_at']);
|
||
});
|
||
|
||
$limit ??= max((int) $this->setting('snapshots.refresh_limit', 2), 1);
|
||
|
||
return array_slice($entries, 0, $limit);
|
||
}
|
||
|
||
private function resolve(string $scope, array $parameters, Closure $resolver, bool $force): array
|
||
{
|
||
$key = $this->cacheKey($scope, $parameters);
|
||
$staleKey = $key . ':stale';
|
||
$queryId = hash('sha256', $scope . '|' . json_encode($parameters, JSON_UNESCAPED_UNICODE));
|
||
$queryLock = $this->cache->lock(
|
||
$this->key('query-lock:' . $queryId),
|
||
max((int) $this->setting('snapshots.lock_seconds', 45), 15)
|
||
);
|
||
|
||
try {
|
||
return $queryLock->block(
|
||
max((int) $this->setting('snapshots.lock_wait_seconds', 3), 1),
|
||
function () use ($scope, $parameters, $resolver, $force, $key, $staleKey, $queryId): array {
|
||
if (! $force) {
|
||
$cached = $this->cache->get($key);
|
||
|
||
if ($this->hasValue($cached)) {
|
||
$this->markRefreshed($scope, $parameters);
|
||
|
||
return ['value' => $cached['value'], 'status' => 'fresh'];
|
||
}
|
||
}
|
||
|
||
if ($this->isCircuitOpen() || $this->cache->has($this->queryCooldownKey($queryId))) {
|
||
return $this->staleOrFail($staleKey, '上游数据正在安全冷却，请稍后再试。');
|
||
}
|
||
|
||
if (! $this->consumeBudget()) {
|
||
return $this->staleOrFail($staleKey, '上游查询刷新已达到安全上限，请稍后再试。');
|
||
}
|
||
|
||
$slots = $this->acquireUpstreamSlots();
|
||
|
||
if ($slots === null) {
|
||
return $this->staleOrFail($staleKey, '上游同步任务繁忙，请稍后再试。');
|
||
}
|
||
|
||
$startedAt = microtime(true);
|
||
|
||
try {
|
||
$value = $resolver();
|
||
} catch (Throwable $exception) {
|
||
$this->recordFailure($scope, $queryId, $exception);
|
||
|
||
return $this->staleOrFail($staleKey, '上游数据暂时不可用，请稍后再试。', $exception);
|
||
} finally {
|
||
$this->releaseUpstreamSlots($slots);
|
||
}
|
||
|
||
$durationMs = (int) round((microtime(true) - $startedAt) * 1000);
|
||
$payload = [
|
||
'value' => $value,
|
||
'cached_at' => now()->toIso8601String(),
|
||
'cached_at_ts' => time(),
|
||
];
|
||
|
||
$this->cache->put($key, $payload, max($this->ttlFor($scope), 60));
|
||
$this->cache->put(
|
||
$staleKey,
|
||
$payload,
|
||
max((int) $this->setting('snapshots.stale_ttl_seconds', 86400), $this->ttlFor($scope))
|
||
);
|
||
|
||
$this->recordSuccess($scope, $parameters, $queryId, $durationMs, $value);
|
||
|
||
return ['value' => $value, 'status' => 'fresh'];
|
||
}
|
||
);
|
||
} catch (LockTimeoutException $exception) {
|
||
return $this->staleOrFail($staleKey, '相同数据正在同步，请稍后再试。', $exception);
|
||
}
|
||
}
|
||
|
||
/** @return array<int,Lock>|null */
|
||
private function acquireUpstreamSlots(): ?array
|
||
{
|
||
$sourceSlotCount = min(max((int) $this->setting('snapshots.max_concurrency', 2), 1), 4);
|
||
$globalSlotCount = min(max((int) config('upstreams.runtime.max_concurrency', 4), 1), 8);
|
||
$waitMs = min(max(
|
||
min(
|
||
(int) $this->setting('snapshots.slot_wait_ms', 1500),
|
||
(int) config('upstreams.runtime.slot_wait_ms', 1500)
|
||
),
|
||
0
|
||
), 5000);
|
||
$deadline = microtime(true) + ($waitMs / 1000);
|
||
$lockSeconds = max(
|
||
(int) $this->setting('connection.connect_timeout', 5)
|
||
+ (int) $this->setting('connection.query_timeout', 15)
|
||
+ 10,
|
||
15
|
||
);
|
||
|
||
do {
|
||
for ($sourceSlot = 0; $sourceSlot < $sourceSlotCount; $sourceSlot++) {
|
||
$sourceLock = $this->cache->lock($this->key('upstream-slot:' . $sourceSlot), $lockSeconds);
|
||
|
||
if (! $sourceLock->get()) {
|
||
continue;
|
||
}
|
||
|
||
for ($globalSlot = 0; $globalSlot < $globalSlotCount; $globalSlot++) {
|
||
$globalLock = $this->cache->lock('upstream:global-slot:' . $globalSlot, $lockSeconds);
|
||
|
||
if ($globalLock->get()) {
|
||
return [$sourceLock, $globalLock];
|
||
}
|
||
}
|
||
|
||
$sourceLock->release();
|
||
}
|
||
|
||
if (microtime(true) >= $deadline) {
|
||
break;
|
||
}
|
||
|
||
usleep(50000);
|
||
} while (true);
|
||
|
||
return null;
|
||
}
|
||
|
||
/** @param array<int,Lock> $slots */
|
||
private function releaseUpstreamSlots(array $slots): void
|
||
{
|
||
foreach (array_reverse($slots) as $slot) {
|
||
$slot->release();
|
||
}
|
||
}
|
||
|
||
private function consumeBudget(): bool
|
||
{
|
||
$window = max((int) $this->setting('snapshots.budget_window_seconds', 60), 60);
|
||
$limit = max((int) $this->setting('snapshots.miss_budget_per_window', 6), 1);
|
||
$bucket = intdiv(time(), $window);
|
||
$key = $this->key('upstream-budget:' . $bucket);
|
||
$lock = $this->cache->lock($this->key('upstream-budget-lock'), 3);
|
||
|
||
try {
|
||
return $lock->block(1, function () use ($key, $limit, $window): bool {
|
||
$count = (int) $this->cache->get($key, 0);
|
||
|
||
if ($count >= $limit) {
|
||
return false;
|
||
}
|
||
|
||
$this->cache->put($key, $count + 1, $window + 60);
|
||
|
||
return true;
|
||
});
|
||
} catch (LockTimeoutException) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
private function recordFailure(string $scope, string $queryId, Throwable $exception): void
|
||
{
|
||
$queryCooldown = max((int) $this->setting('snapshots.query_failure_cooldown_seconds', 300), 60);
|
||
$this->cache->put($this->queryCooldownKey($queryId), true, $queryCooldown);
|
||
|
||
if ($exception instanceof ServiceUnavailableHttpException) {
|
||
$globalCooldown = max((int) $this->setting('snapshots.failure_cooldown_seconds', 300), 60);
|
||
$this->cache->put($this->circuitKey(), true, $globalCooldown);
|
||
}
|
||
|
||
Log::warning('上游查询失败，已进入冷却。', [
|
||
'source' => $this->source,
|
||
'source_label' => $this->label,
|
||
'scope' => $scope,
|
||
'exception' => $exception::class,
|
||
]);
|
||
}
|
||
|
||
private function recordSuccess(
|
||
string $scope,
|
||
array $parameters,
|
||
string $queryId,
|
||
int $durationMs,
|
||
mixed $value
|
||
): void {
|
||
$slowMs = max((int) $this->setting('snapshots.slow_query_ms', 5000), 1000);
|
||
|
||
if ($durationMs >= $slowMs) {
|
||
$cooldown = max((int) $this->setting('snapshots.slow_query_cooldown_seconds', 300), 60);
|
||
$this->cache->put($this->queryCooldownKey($queryId), true, $cooldown);
|
||
Log::warning('上游查询较慢，已降低该查询刷新频率。', [
|
||
'source' => $this->source,
|
||
'source_label' => $this->label,
|
||
'scope' => $scope,
|
||
'duration_ms' => $durationMs,
|
||
]);
|
||
}
|
||
|
||
$this->markRefreshed($scope, $parameters, [
|
||
'last_duration_ms' => $durationMs,
|
||
'last_row_count' => $this->rowCount($value),
|
||
'last_refreshed_at' => time(),
|
||
]);
|
||
}
|
||
|
||
private function register(string $scope, array $parameters, bool $urgent): void
|
||
{
|
||
$entryKey = $this->entryKey($scope, $parameters);
|
||
$touchSeconds = $urgent ? 30 : 300;
|
||
|
||
if (! $this->cache->add($this->key('registry-touch:' . $entryKey . ':' . (int) $urgent), true, $touchSeconds)) {
|
||
return;
|
||
}
|
||
|
||
$this->updateRegistry(function (array $entries) use ($entryKey, $scope, $parameters, $urgent): array {
|
||
$existing = (array) ($entries[$entryKey] ?? []);
|
||
$entries[$entryKey] = array_merge($existing, [
|
||
'source' => $this->source,
|
||
'scope' => $scope,
|
||
'parameters' => $parameters,
|
||
'seen_at' => time(),
|
||
'urgent' => $urgent || (bool) ($existing['urgent'] ?? false),
|
||
]);
|
||
|
||
return $entries;
|
||
});
|
||
}
|
||
|
||
private function markRefreshed(string $scope, array $parameters, array $metrics = []): void
|
||
{
|
||
$entryKey = $this->entryKey($scope, $parameters);
|
||
|
||
$this->updateRegistry(function (array $entries) use ($entryKey, $metrics): array {
|
||
if (! isset($entries[$entryKey])) {
|
||
return $entries;
|
||
}
|
||
|
||
$entries[$entryKey] = array_merge($entries[$entryKey], $metrics, ['urgent' => false]);
|
||
|
||
return $entries;
|
||
});
|
||
}
|
||
|
||
private function updateRegistry(Closure $callback): void
|
||
{
|
||
$lock = $this->cache->lock($this->key('registry-lock:' . self::CACHE_VERSION), 5);
|
||
|
||
try {
|
||
$lock->block(2, function () use ($callback): void {
|
||
$entries = $this->cache->get($this->registryKey(), []);
|
||
$entries = is_array($entries) ? $entries : [];
|
||
$entries = $callback($entries);
|
||
|
||
uasort($entries, static fn (array $a, array $b): int => ((int) $b['seen_at']) <=> ((int) $a['seen_at']));
|
||
$limit = max((int) $this->setting('snapshots.registry_limit', 300), 1);
|
||
|
||
$this->cache->forever($this->registryKey(), array_slice($entries, 0, $limit, true));
|
||
});
|
||
} catch (LockTimeoutException) {
|
||
// 登记失败不能阻塞页面；下一次访问或刷新会再次尝试。
|
||
}
|
||
}
|
||
|
||
private function staleOrFail(string $staleKey, string $message, ?Throwable $previous = null): array
|
||
{
|
||
$stale = $this->cache->get($staleKey);
|
||
|
||
if ($this->hasValue($stale)) {
|
||
return ['value' => $stale['value'], 'status' => 'stale'];
|
||
}
|
||
|
||
throw new ServiceUnavailableHttpException(30, $message, $previous);
|
||
}
|
||
|
||
private function isCircuitOpen(): bool
|
||
{
|
||
return $this->cache->has($this->circuitKey());
|
||
}
|
||
|
||
private function queryCooldownKey(string $queryId): string
|
||
{
|
||
return $this->key('query-cooldown:' . $queryId);
|
||
}
|
||
|
||
private function isScopeAllowed(string $scope): bool
|
||
{
|
||
$allowed = (array) $this->setting('allowed_scopes', []);
|
||
|
||
foreach ($allowed as $pattern) {
|
||
$pattern = trim((string) $pattern);
|
||
|
||
if ($pattern === '*' || $pattern === $scope) {
|
||
return true;
|
||
}
|
||
|
||
if (str_ends_with($pattern, '.*') && str_starts_with($scope, substr($pattern, 0, -1))) {
|
||
return true;
|
||
}
|
||
}
|
||
|
||
return false;
|
||
}
|
||
|
||
private function usesDirectDevelopmentRead(): bool
|
||
{
|
||
return (bool) config('upstreams.development.direct_read', false);
|
||
}
|
||
|
||
private function entryKey(string $scope, array $parameters): string
|
||
{
|
||
return hash('sha256', $scope . '|' . json_encode($parameters, JSON_UNESCAPED_UNICODE));
|
||
}
|
||
|
||
private function cacheKey(string $scope, array $parameters): string
|
||
{
|
||
return $this->key('snapshot:' . self::CACHE_VERSION . ':' . $scope . ':'
|
||
. hash('sha256', json_encode($parameters, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION)));
|
||
}
|
||
|
||
private function ttlFor(string $scope): int
|
||
{
|
||
if (str_ends_with($scope, '.options')) {
|
||
return (int) $this->setting('snapshots.options_ttl_seconds', 3600);
|
||
}
|
||
|
||
if (str_ends_with($scope, '.summary')) {
|
||
return (int) $this->setting('snapshots.summary_ttl_seconds', 120);
|
||
}
|
||
|
||
return (int) $this->setting('snapshots.ttl_seconds', 120);
|
||
}
|
||
|
||
private function isDue(array $entry, int $now): bool
|
||
{
|
||
if ((bool) ($entry['urgent'] ?? false)) {
|
||
return true;
|
||
}
|
||
|
||
$scope = (string) $entry['scope'];
|
||
$parameters = (array) $entry['parameters'];
|
||
$cached = $this->cache->get($this->cacheKey($scope, $parameters));
|
||
|
||
if (! is_array($cached) || ! isset($cached['cached_at_ts'])) {
|
||
return true;
|
||
}
|
||
|
||
return ((int) $cached['cached_at_ts'] + max($this->ttlFor($scope), 60)) <= ($now + 15);
|
||
}
|
||
|
||
private function priorityFor(string $scope): int
|
||
{
|
||
return match (true) {
|
||
str_ends_with($scope, '.list') => 3,
|
||
str_ends_with($scope, '.summary') => 2,
|
||
default => 1,
|
||
};
|
||
}
|
||
|
||
private function rowCount(mixed $value): int
|
||
{
|
||
if (! is_array($value)) {
|
||
return 1;
|
||
}
|
||
|
||
if (isset($value['data']) && is_array($value['data'])) {
|
||
return count($value['data']);
|
||
}
|
||
|
||
return count($value);
|
||
}
|
||
|
||
private function hasValue(mixed $payload): bool
|
||
{
|
||
return is_array($payload) && array_key_exists('value', $payload);
|
||
}
|
||
|
||
private function setReadState(string $status, array $payload): void
|
||
{
|
||
$this->lastReadState = [
|
||
'status' => $status,
|
||
'cached_at' => $payload['cached_at'] ?? null,
|
||
];
|
||
}
|
||
|
||
private function normalize(array $parameters): array
|
||
{
|
||
foreach ($parameters as &$value) {
|
||
if (is_array($value)) {
|
||
$value = $this->normalize($value);
|
||
}
|
||
}
|
||
unset($value);
|
||
|
||
ksort($parameters);
|
||
|
||
return $parameters;
|
||
}
|
||
|
||
/**
|
||
* 快照必须按公司和权限范围隔离。后台刷新沿用首次访问时登记的上下文，
|
||
* 不依赖无登录用户的 CLI 进程重新推断权限。
|
||
*/
|
||
private function prepareParameters(array $parameters): array
|
||
{
|
||
$contextKey = '__caiwu_snapshot_context';
|
||
$context = isset($parameters[$contextKey]) && is_array($parameters[$contextKey])
|
||
? $this->normalizeContext($parameters[$contextKey])
|
||
: $this->currentAccessContext();
|
||
|
||
unset($parameters[$contextKey]);
|
||
$parameters[$contextKey] = $context;
|
||
|
||
return $this->normalize($parameters);
|
||
}
|
||
|
||
private function currentAccessContext(): array
|
||
{
|
||
if ($this->accessContext !== null) {
|
||
return $this->accessContext;
|
||
}
|
||
|
||
$companyId = function_exists('company_id') ? (int) (company_id() ?? 0) : 0;
|
||
$user = auth()->user();
|
||
|
||
if ($user === null) {
|
||
return $this->accessContext = [
|
||
'company_id' => $companyId,
|
||
'access_hash' => 'system',
|
||
];
|
||
}
|
||
|
||
$permissionIds = $user->allPermissions(['id'])
|
||
->pluck('id')
|
||
->map(static fn (mixed $id): int => (int) $id)
|
||
->sort()
|
||
->values()
|
||
->all();
|
||
|
||
return $this->accessContext = [
|
||
'company_id' => $companyId,
|
||
'access_hash' => hash('sha256', json_encode(
|
||
$permissionIds,
|
||
JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION
|
||
)),
|
||
];
|
||
}
|
||
|
||
private function normalizeContext(array $context): array
|
||
{
|
||
return [
|
||
'company_id' => max((int) ($context['company_id'] ?? 0), 0),
|
||
'access_hash' => preg_match('/^[a-f0-9]{64}$/', (string) ($context['access_hash'] ?? '')) === 1
|
||
? (string) $context['access_hash']
|
||
: 'system',
|
||
];
|
||
}
|
||
|
||
private function key(string $suffix): string
|
||
{
|
||
return $this->source . ':' . $suffix;
|
||
}
|
||
|
||
private function registryKey(): string
|
||
{
|
||
return $this->key('snapshots:active:' . self::CACHE_VERSION);
|
||
}
|
||
|
||
private function circuitKey(): string
|
||
{
|
||
return $this->key('upstream-circuit:' . self::CACHE_VERSION);
|
||
}
|
||
}
|