<?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;
            }

            $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);
    }
}