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 */ public function consumeSince(int $afterId): array { return $this->consumeInboxSince($this->inboxKey(), $afterId); } /** * @return array */ public function consumeCustomerSince(int $customerId, int $afterId): array { if ($customerId <= 0) { return []; } return $this->consumeInboxSince($this->customerInboxKey($customerId), $afterId); } /** * @param array|array $data */ public function publish(string $event, array $data): void { if (! $this->isEnabled()) { return; } $this->publishToInbox($this->inboxKey(), $event, $data); } /** * @param array|array $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 */ 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|array $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|array $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)); } }