From 17d79c0ce680c13272a02d5e5c5b79b8d262d8e8 Mon Sep 17 00:00:00 2001 From: adonis Date: Thu, 27 Aug 2026 11:53:29 +0330 Subject: [PATCH] fix some issue from sse --- .../Site/PriceStreamController.php | 75 +++++++++++++++---- app/Services/PriceStreamService.php | 39 ++++++++-- 2 files changed, 95 insertions(+), 19 deletions(-) diff --git a/app/Http/Controllers/Site/PriceStreamController.php b/app/Http/Controllers/Site/PriceStreamController.php index e2943d1..0f37d25 100644 --- a/app/Http/Controllers/Site/PriceStreamController.php +++ b/app/Http/Controllers/Site/PriceStreamController.php @@ -53,33 +53,48 @@ class PriceStreamController extends Controller $cache = app(LivePriceCacheService::class); $config = app(\App\Services\ConfigService::class); - $last = $cache->get(); - // کش ممکن است no_quote صفرِ کوتاه‌عمر داشته باشد؛ با build از montzar_live_price جبران کن - if (! is_array($last) || ! $config->payloadHasDisplayablePrice($last)) { - $last = $cache->buildFromDatabase(); - } - if (is_array($last) && $last !== []) { - if ($customerId > 0) { - $last = app(\App\Services\CustomerSpecialPricingService::class) - ->applyToLastPayload($last, $customerId); + $specialPricing = app(\App\Services\CustomerSpecialPricingService::class); + $zaarbox = app(ZaarboxPriceService::class); + $broadcast = app(PriceBroadcastService::class); + + $resolveLastPayload = static function () use ($cache, $config, $specialPricing, $zaarbox, $customerId): ?array { + $last = $cache->get(); + if (! is_array($last) || ! $config->payloadHasDisplayablePrice($last)) { + $last = $cache->buildFromDatabase(); } - $send('last', app(ZaarboxPriceService::class)->attachGoldOunceToPayload($last)); + if (! is_array($last) || $last === []) { + return null; + } + if ($customerId > 0) { + $last = $specialPricing->applyToLastPayload($last, $customerId); + } + + return $zaarbox->attachGoldOunceToPayload($last); + }; + + $last = $resolveLastPayload(); + $lastSentIdentity = null; + if (is_array($last) && $last !== []) { + $send('last', $last); + $lastSentIdentity = PriceBroadcastService::lastPriceIdentity($last); } - $hoursSettings = app(\App\Services\ConfigService::class)->clientTradingHoursSettings(); + $hoursSettings = $config->clientTradingHoursSettings(); if ($hoursSettings !== []) { $send('settings', $hoursSettings); } - $list = app(PriceBroadcastService::class)->buildListPayload(false); + $list = $broadcast->buildListPayload(false); + $lastSentListIdentity = null; if (is_array($list) && $list !== []) { $send('list', $list); + $lastSentListIdentity = PriceBroadcastService::listIdentity($list); } if ($customerId > 0) { $send( 'special-pricing', - app(\App\Services\CustomerSpecialPricingService::class)->clientPayloadForCustomer($customerId) + $specialPricing->clientPayloadForCustomer($customerId) ); } @@ -109,8 +124,16 @@ class PriceStreamController extends Controller while (! connection_aborted()) { foreach ($service->consumeSince($lastId) as $event) { - $send((string) $event['event'], $event['data']); + $eventName = (string) $event['event']; + $eventData = $event['data']; + $send($eventName, $eventData); $lastId = (int) $event['id']; + if ($eventName === 'last' && is_array($eventData)) { + $lastSentIdentity = PriceBroadcastService::lastPriceIdentity($eventData); + } + if ($eventName === 'list' && is_array($eventData)) { + $lastSentListIdentity = PriceBroadcastService::listIdentity($eventData); + } } if ($customerId > 0) { @@ -121,12 +144,36 @@ class PriceStreamController extends Controller } $heartbeatTicks++; + + // هر ۱ ثانیه: قیمت را مستقیم از کش/DB بکش + // (اگر inbox بین tick و PHP-FPM هم‌رسانی نشود، فقط ping دیده می‌شد) + if ($heartbeatTicks % 2 === 0) { + $pulled = $resolveLastPayload(); + if (is_array($pulled) && $config->payloadHasDisplayablePrice($pulled)) { + $identity = PriceBroadcastService::lastPriceIdentity($pulled); + if ($lastSentIdentity !== $identity) { + $send('last', $pulled); + $lastSentIdentity = $identity; + } + } + } + // هر ۵ ثانیه — LiteSpeed/پروکسی اتصال بیکار را نمی‌بندد if ($heartbeatTicks >= 10) { $send('ping', ['ts' => time()]); if ($customerId > 0) { $presence->touch($customerId); } + + $pulledList = $broadcast->buildListPayload(true); + if (is_array($pulledList) && $pulledList !== []) { + $listIdentity = PriceBroadcastService::listIdentity($pulledList); + if ($lastSentListIdentity !== $listIdentity) { + $send('list', $pulledList); + $lastSentListIdentity = $listIdentity; + } + } + $heartbeatTicks = 0; } diff --git a/app/Services/PriceStreamService.php b/app/Services/PriceStreamService.php index c728ac1..df57b6c 100644 --- a/app/Services/PriceStreamService.php +++ b/app/Services/PriceStreamService.php @@ -2,8 +2,10 @@ namespace App\Services; +use Illuminate\Contracts\Cache\Repository; use Illuminate\Http\Request; use Illuminate\Support\Facades\Cache; +use Illuminate\Support\Facades\Log; class PriceStreamService { @@ -99,6 +101,31 @@ class PriceStreamService $this->publishToInbox($this->customerInboxKey($customerId), $event, $data); } + /** + * Inbox باید بین tick (CLI) و SSE (PHP-FPM) مشترک باشد. + * file cache روی چند ورکر/سرور inbox را جدا می‌کند → فقط ping دیده می‌شود. + */ + private function store(): Repository + { + static $store = null; + if ($store instanceof Repository) { + return $store; + } + + try { + $redis = Cache::store('redis'); + $redis->get('price_sse_inbox:ping'); + $store = $redis; + } catch (\Throwable $e) { + Log::warning('PriceStreamService: redis unavailable, falling back to default cache', [ + 'message' => $e->getMessage(), + ]); + $store = Cache::store(); + } + + return $store; + } + private function inboxKey(): string { return self::INBOX_PREFIX.'global'; @@ -111,7 +138,7 @@ class PriceStreamService private function currentInboxSeq(string $key): int { - $inbox = Cache::get($key, ['seq' => 0, 'events' => []]); + $inbox = $this->store()->get($key, ['seq' => 0, 'events' => []]); return (int) ($inbox['seq'] ?? 0); } @@ -121,7 +148,7 @@ class PriceStreamService */ private function consumeInboxSince(string $key, int $afterId): array { - $inbox = Cache::get($key, ['seq' => 0, 'events' => []]); + $inbox = $this->store()->get($key, ['seq' => 0, 'events' => []]); $events = is_array($inbox['events'] ?? null) ? $inbox['events'] : []; return array_values(array_filter( @@ -135,8 +162,10 @@ class PriceStreamService */ private function publishToInbox(string $key, string $event, array $data): void { - Cache::lock($key.':lock', 5)->block(3, function () use ($key, $event, $data): void { - $inbox = Cache::get($key, ['seq' => 0, 'events' => []]); + $store = $this->store(); + + $store->lock($key.':lock', 5)->block(3, function () use ($store, $key, $event, $data): void { + $inbox = $store->get($key, ['seq' => 0, 'events' => []]); $events = is_array($inbox['events'] ?? null) ? $inbox['events'] : []; $lastEvent = $events !== [] ? $events[array_key_last($events)] : null; @@ -157,7 +186,7 @@ class PriceStreamService $inbox['events'] = array_slice($inbox['events'], -self::MAX_EVENTS); } - Cache::put($key, $inbox, now()->addSeconds(self::INBOX_TTL_SECONDS)); + $store->put($key, $inbox, now()->addSeconds(self::INBOX_TTL_SECONDS)); }); }