boursecu_server/app/Http/Controllers/Site/PriceStreamController.php

<?php

namespace App\Http\Controllers\Site;

use App\Http\Controllers\Controller;
use App\Services\AssetSummaryService;
use App\Services\CustomerPresenceService;
use App\Services\LivePriceCacheService;
use App\Services\PriceBroadcastService;
use App\Services\PriceStreamService;
use App\Services\ZaarboxPriceService;
use Illuminate\Http\Request;
use Symfony\Component\HttpFoundation\StreamedResponse;

class PriceStreamController extends Controller
{
    public function stream(Request $request): StreamedResponse
    {
        $service = app(PriceStreamService::class);

        if (! $service->isEnabled()) {
            abort(404);
        }

        $customer = auth('customer')->user();
        $customerId = $customer ? (int) $customer->id : 0;
        $customerOwnerType = $customer ? (int) $customer->type : 0;

        return response()->stream(function () use ($service, $customerId, $customerOwnerType) {
            @set_time_limit(0);

            if (function_exists('apache_setenv')) {
                @apache_setenv('no-gzip', '1');
            }

            @ini_set('zlib.output_compression', '0');
            @ini_set('implicit_flush', '1');

            while (ob_get_level() > 0) {
                @ob_end_flush();
            }

            $send = static function (string $event, $data): void {
                echo 'event: '.$event."\n";
                echo 'data: '.json_encode($data, JSON_UNESCAPED_UNICODE)."\n\n";
                @flush();
            };

            $presence = app(CustomerPresenceService::class);
            if ($customerId > 0) {
                $presence->touch($customerId, true);
            }

            $cache = app(LivePriceCacheService::class);
            $config = app(\App\Services\ConfigService::class);
            $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();
                }
                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 = $config->clientTradingHoursSettings();
            if ($hoursSettings !== []) {
                $send('settings', $hoursSettings);
            }

            $list = $broadcast->buildListPayload(false);
            $lastSentListIdentity = null;
            if (is_array($list) && $list !== []) {
                $send('list', $list);
                $lastSentListIdentity = PriceBroadcastService::listIdentity($list);
            }

            if ($customerId > 0) {
                $send(
                    'special-pricing',
                    $specialPricing->clientPayloadForCustomer($customerId)
                );
            }

            $customerLastId = 0;
            if ($customerId > 0) {
                $assets = app(AssetSummaryService::class)->buildForCustomer(
                    $customerId,
                    $customerOwnerType > 0 ? $customerOwnerType : 2
                );
                if ($assets !== []) {
                    $send('assets', $assets);
                }

                $unread = app(\App\Services\AppNotificationUnreadService::class);
                if ($unread->featureEnabled()) {
                    $send('unread-count', $unread->payloadForCustomer($customerId));
                }

                $customerLastId = $service->currentCustomerSeq($customerId);
            }

            $lastId = $service->currentSeq();
            $heartbeatTicks = 0;

            echo "retry: 5000\n\n";
            @flush();

            while (! connection_aborted()) {
                foreach ($service->consumeSince($lastId) as $event) {
                    $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) {
                    foreach ($service->consumeCustomerSince($customerId, $customerLastId) as $event) {
                        $send((string) $event['event'], $event['data']);
                        $customerLastId = (int) $event['id'];
                    }
                }

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

                usleep(500000);
            }
        }, 200, [
            'Content-Type' => 'text/event-stream; charset=UTF-8',
            'Cache-Control' => 'no-cache, no-store, must-revalidate',
            'Connection' => 'keep-alive',
            'X-Accel-Buffering' => 'no',
        ]);
    }
}