boursecu_server/app/Services/PriceStreamService.php

<?php

namespace App\Services;

use Illuminate\Contracts\Cache\Repository;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Log;

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
    {
        return app(ConfigService::class)->resolvePublicHost($request);
    }

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

    /**
     * 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;
    }

    /**
     * چند مستأجر روی یک Redis با پیشوند یکسان inbox را قاطی می‌کردند
     * (لیست شمش یک فروشگاه روی صفحهٔ فروشگاه دیگر).
     */
    public function tenantScope(): string
    {
        $default = (string) config('database.default', 'mysql');
        $db = (string) config("database.connections.{$default}.database");
        $host = '';
        try {
            $host = app(ConfigService::class)->resolvePublicHost();
        } catch (\Throwable) {
            $host = '';
        }

        $parts = array_values(array_filter(
            [$db, $host],
            static fn (string $value): bool => $value !== ''
        ));

        return $parts !== [] ? implode(':', $parts) : 'default';
    }

    private function inboxKey(): string
    {
        return self::INBOX_PREFIX.$this->tenantScope();
    }

    private function customerInboxKey(int $customerId): string
    {
        return self::CUSTOMER_INBOX_PREFIX.$this->tenantScope().':'.$customerId;
    }

    private function currentInboxSeq(string $key): int
    {
        $inbox = $this->store()->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 = $this->store()->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
    {
        $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;

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

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