boursecu_server/app/Services/PriceStreamService.php

<?php

namespace App\Services;

use Illuminate\Http\Request;
use Illuminate\Support\Facades\Cache;

class PriceStreamService
{
    private const INBOX_PREFIX = 'price_sse_inbox:';

    private const CUSTOMER_INBOX_PREFIX = 'price_sse_inbox:customer:';

    private const INBOX_TTL_SECONDS = 300;

    private const MAX_EVENTS = 50;

    public function isEnabled(): bool
    {
        return app(ConfigService::class)->getBool('wss_enabled', true);
    }

    public function streamPath(): string
    {
        $config = app(ConfigService::class);
        $path = trim($config->getString('sse_path', ''));

        if ($path === '') {
            $path = trim($config->getString('wss_path', ''));
        }

        if ($path === '' || strpos($path, '/prc/') !== false) {
            return '/api/price/stream';
        }

        return strpos($path, '/') === 0 ? $path : "/{$path}";
    }

    public function resolveHost(?Request $request = null): string
    {
        $config = app(ConfigService::class);

        $appUrl = trim($config->getString('app_url', ''));
        if ($appUrl !== '') {
            $host = parse_url($appUrl, PHP_URL_HOST);
            if (is_string($host) && $host !== '') {
                return $this->normalizeHost($host);
            }
        }

        if ($request !== null) {
            $requestHost = trim((string) $request->getHost());
            if ($requestHost !== '') {
                return $this->normalizeHost($requestHost);
            }
        }

        if (! app()->runningInConsole() && request()) {
            $requestHost = trim((string) request()->getHost());
            if ($requestHost !== '') {
                return $this->normalizeHost($requestHost);
            }
        }

        $nameEn = trim((string) $config->getString('name_en', ''));
        if ($nameEn !== '') {
            return $this->normalizeHost($nameEn);
        }

        return '';
    }

    public function currentSeq(): int
    {
        return $this->currentInboxSeq($this->inboxKey());
    }

    public function currentCustomerSeq(int $customerId): int
    {
        if ($customerId <= 0) {
            return 0;
        }

        return $this->currentInboxSeq($this->customerInboxKey($customerId));
    }

    /**
     * @return array<int, array{id: int, event: string, data: mixed}>
     */
    public function consumeSince(int $afterId): array
    {
        return $this->consumeInboxSince($this->inboxKey(), $afterId);
    }

    /**
     * @return array<int, array{id: int, event: string, data: mixed}>
     */
    public function consumeCustomerSince(int $customerId, int $afterId): array
    {
        if ($customerId <= 0) {
            return [];
        }

        return $this->consumeInboxSince($this->customerInboxKey($customerId), $afterId);
    }

    /**
     * @param array<string, mixed>|array<int, mixed> $data
     */
    public function publish(string $event, array $data): void
    {
        if (! $this->isEnabled()) {
            return;
        }

        $this->publishToInbox($this->inboxKey(), $event, $data);
    }

    /**
     * @param array<string, mixed>|array<int, mixed> $data
     */
    public function publishForCustomer(int $customerId, string $event, array $data): void
    {
        if (! $this->isEnabled() || $customerId <= 0) {
            return;
        }

        $this->publishToInbox($this->customerInboxKey($customerId), $event, $data);
    }

    private function inboxKey(): string
    {
        return self::INBOX_PREFIX.'global';
    }

    private function customerInboxKey(int $customerId): string
    {
        return self::CUSTOMER_INBOX_PREFIX.$customerId;
    }

    private function currentInboxSeq(string $key): int
    {
        $inbox = Cache::get($key, ['seq' => 0, 'events' => []]);

        return (int) ($inbox['seq'] ?? 0);
    }

    /**
     * @return array<int, array{id: int, event: string, data: mixed}>
     */
    private function consumeInboxSince(string $key, int $afterId): array
    {
        $inbox = Cache::get($key, ['seq' => 0, 'events' => []]);
        $events = is_array($inbox['events'] ?? null) ? $inbox['events'] : [];

        return array_values(array_filter(
            $events,
            static fn (array $event): bool => (int) ($event['id'] ?? 0) > $afterId
        ));
    }

    /**
     * @param array<string, mixed>|array<int, mixed> $data
     */
    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' => []]);
            $events = is_array($inbox['events'] ?? null) ? $inbox['events'] : [];
            $lastEvent = $events !== [] ? $events[array_key_last($events)] : null;

            if (is_array($lastEvent)
                && ($lastEvent['event'] ?? null) === $event
                && $this->eventDataHash($event, $lastEvent['data'] ?? null) === $this->eventDataHash($event, $data)) {
                return;
            }

            $inbox['seq'] = (int) ($inbox['seq'] ?? 0) + 1;
            $inbox['events'][] = [
                'id' => $inbox['seq'],
                'event' => $event,
                'data' => $data,
            ];

            if (count($inbox['events']) > self::MAX_EVENTS) {
                $inbox['events'] = array_slice($inbox['events'], -self::MAX_EVENTS);
            }

            Cache::put($key, $inbox, now()->addSeconds(self::INBOX_TTL_SECONDS));
        });
    }

    /**
     * @param array<string, mixed>|array<int, mixed> $data
     */
    private function eventDataHash(string $event, mixed $data): string
    {
        if (! is_array($data)) {
            return md5('');
        }

        if ($event === 'last') {
            return md5(json_encode(PriceBroadcastService::lastPriceIdentity($data), JSON_UNESCAPED_UNICODE));
        }

        if ($event === 'list') {
            return md5(json_encode(PriceBroadcastService::listIdentity($data), JSON_UNESCAPED_UNICODE));
        }

        if ($event === 'assets') {
            return md5(json_encode([
                'cash_wallet' => (int) ($data['cash_wallet'] ?? 0),
                'total_rial_value' => (int) ($data['total_rial_value'] ?? 0),
                'assets_rial_value' => (int) ($data['assets_rial_value'] ?? 0),
                'inventory' => md5(json_encode($data['inventory_cards'] ?? [], JSON_UNESCAPED_UNICODE)),
                'metals' => md5(json_encode($data['metal_totals'] ?? [], JSON_UNESCAPED_UNICODE)),
            ], JSON_UNESCAPED_UNICODE));
        }

        return md5(json_encode($data, JSON_UNESCAPED_UNICODE));
    }

    private function normalizeHost(string $host): string
    {
        $host = strtolower(trim($host));

        return strpos($host, 'www.') === 0 ? substr($host, 4) : $host;
    }
}