From f44063dbd29cd826f8f8c759efac3db9ba1e997b Mon Sep 17 00:00:00 2001 From: adonis Date: Thu, 18 Jun 2026 16:12:53 +0330 Subject: [PATCH] added wss tools 4 --- .../Site/PriceStreamController.php | 76 ++++++++++ app/Services/PriceBroadcastService.php | 37 +++-- app/Services/PriceStreamService.php | 133 ++++++++++++++++++ database/sql/deploy_settings_const.sql | 28 ++-- database/sql/wss_settings_update.sql | 48 +++---- routes/api.php | 2 + 6 files changed, 268 insertions(+), 56 deletions(-) create mode 100644 app/Http/Controllers/Site/PriceStreamController.php create mode 100644 app/Services/PriceStreamService.php diff --git a/app/Http/Controllers/Site/PriceStreamController.php b/app/Http/Controllers/Site/PriceStreamController.php new file mode 100644 index 0000000..ec056d0 --- /dev/null +++ b/app/Http/Controllers/Site/PriceStreamController.php @@ -0,0 +1,76 @@ +isEnabled()) { + abort(404); + } + + $host = $service->resolveHost($request); + + return response()->stream(function () use ($service, $host) { + @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(); + }; + + $cache = app(LivePriceCacheService::class); + $last = $cache->get() ?? $cache->buildFromDatabase(); + if (is_array($last) && $last !== []) { + $send('last', $last); + } + + $lastId = $host !== '' ? $service->currentSeq($host) : 0; + $heartbeatTicks = 0; + + while (! connection_aborted()) { + if ($host !== '') { + foreach ($service->consumeSince($host, $lastId) as $event) { + $send((string) $event['event'], $event['data']); + $lastId = (int) $event['id']; + } + } + + $heartbeatTicks++; + if ($heartbeatTicks >= 30) { + echo ': ping '.time()."\n\n"; + @flush(); + $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', + ]); + } +} diff --git a/app/Services/PriceBroadcastService.php b/app/Services/PriceBroadcastService.php index 92ebccc..a8a0c99 100644 --- a/app/Services/PriceBroadcastService.php +++ b/app/Services/PriceBroadcastService.php @@ -8,25 +8,19 @@ use Illuminate\Support\Facades\Cache; class PriceBroadcastService { - private const LAST_HASH_KEY = 'ws_broadcast:last_hash'; + private const LAST_HASH_KEY = 'price_broadcast:last_hash'; - private const LIST_HASH_KEY = 'ws_broadcast:list_hash'; + private const LIST_HASH_KEY = 'price_broadcast:list_hash'; /** - * montzar فقط trigger «price_update» را به کلاینت‌ها relay می‌کند. + * برادکست SSE — payload کامل last/list وقتی داده عوض شود. * * @param array|null $lastPayload */ public function broadcastAfterPriceUpdate(?array $lastPayload = null, bool $force = false): void { - $ws = app(WebsocketBroadcastService::class); - if (! $ws->isEnabled()) { - return; - } - - if ($force) { - $ws->sendEvent('price_update', []); - + $stream = app(PriceStreamService::class); + if (! $stream->isEnabled()) { return; } @@ -35,6 +29,19 @@ class PriceBroadcastService $lastPayload = $cache->get() ?? $cache->buildFromDatabase(); } + if ($force) { + if (is_array($lastPayload) && $lastPayload !== []) { + $stream->publish('last', $lastPayload); + } + + $listPayload = $this->buildProductListPayload(); + if ($listPayload !== null) { + $stream->publish('list', $listPayload); + } + + return; + } + $lastChanged = is_array($lastPayload) && $lastPayload !== [] && $this->rememberHashIfChanged(self::LAST_HASH_KEY, $lastPayload); @@ -42,8 +49,12 @@ class PriceBroadcastService $listChanged = $listPayload !== null && $this->rememberHashIfChanged(self::LIST_HASH_KEY, $listPayload); - if ($lastChanged || $listChanged) { - $ws->sendEvent('price_update', []); + if ($lastChanged) { + $stream->publish('last', $lastPayload); + } + + if ($listChanged) { + $stream->publish('list', $listPayload); } } diff --git a/app/Services/PriceStreamService.php b/app/Services/PriceStreamService.php new file mode 100644 index 0000000..c12e17e --- /dev/null +++ b/app/Services/PriceStreamService.php @@ -0,0 +1,133 @@ +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(string $host): int + { + $inbox = Cache::get($this->inboxKey($host), ['seq' => 0, 'events' => []]); + + return (int) ($inbox['seq'] ?? 0); + } + + /** + * @return array + */ + public function consumeSince(string $host, int $afterId): array + { + $inbox = Cache::get($this->inboxKey($host), ['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 + */ + public function publish(string $event, array $data, ?string $host = null): void + { + if (! $this->isEnabled()) { + return; + } + + $host = $host ?? $this->resolveHost(); + if ($host === '') { + return; + } + + $key = $this->inboxKey($host); + $inbox = Cache::get($key, ['seq' => 0, 'events' => []]); + $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)); + } + + private function inboxKey(string $host): string + { + return self::INBOX_PREFIX.$this->normalizeHost($host); + } + + private function normalizeHost(string $host): string + { + $host = strtolower(trim($host)); + + return strpos($host, 'www.') === 0 ? substr($host, 4) : $host; + } +} diff --git a/database/sql/deploy_settings_const.sql b/database/sql/deploy_settings_const.sql index faedd01..420a60f 100644 --- a/database/sql/deploy_settings_const.sql +++ b/database/sql/deploy_settings_const.sql @@ -205,15 +205,17 @@ INSERT INTO `settings` (`key`, `value`, `title_fa`, `can_edit`, `is_active`, `de SELECT 'external_order_api_enabled', 'true', 'فعال‌سازی API سفارش خارجی (get_wallet / order)', 1, 1, 'true + node_client_id + node_shared_secret' WHERE NOT EXISTS (SELECT 1 FROM `settings` WHERE `key` = 'external_order_api_enabled' LIMIT 1); --- WebSocket قیمت (برادکست last/list) --- هر پروژه دو پورت host: WS_SOCKET_PORT→8080 (لایت‌اسپید /prc/ws) و WS_HTTP_PORT→3000 (لاراول trigger) --- مثال طباطبایی: socket=2011، http=2012 — مثال رحیمی: socket=2010، http=2009 (هر پروژه جفت خودش) +-- WebSocket قیمت (SSE — last/list از لاراول) INSERT INTO `settings` (`key`, `value`, `title_fa`, `can_edit`, `is_active`, `description`, `type`) -SELECT 'wss_enabled', 'true', 'قیمت لحظه‌ای از WebSocket', 0, 1, 'روشن: اپ از wss_path استفاده می‌کند؛ خاموش: polling /api/price/last و /customer/product/list', 'switch' +SELECT 'wss_enabled', 'true', 'قیمت لحظه‌ای (SSE)', 0, 1, 'روشن: اپ به sse_path وصل می‌شود؛ خاموش: polling /api/price/last و /customer/product/list', 'switch' WHERE NOT EXISTS (SELECT 1 FROM `settings` WHERE `key` = 'wss_enabled' LIMIT 1); INSERT INTO `settings` (`key`, `value`, `title_fa`, `can_edit`, `is_active`, `description`, `type`) -SELECT 'wss_path', '/prc/ws', 'مسیر WebSocket قیمت', 0, 1, 'مسیر روی دامنه اپ — اپ به wss://{دامنه}{این_مسیر} وصل می‌شود. لایت‌اسپید WS Proxy همین مسیر را به wss_socket_port می‌فرستد.', 'text' +SELECT 'sse_path', '/api/price/stream', 'مسیر SSE قیمت', 0, 1, 'GET لاراول — استریم event:last و event:list با payload کامل', 'text' +WHERE NOT EXISTS (SELECT 1 FROM `settings` WHERE `key` = 'sse_path' LIMIT 1); + +INSERT INTO `settings` (`key`, `value`, `title_fa`, `can_edit`, `is_active`, `description`, `type`) +SELECT 'wss_path', '/prc/ws', 'مسیر WebSocket (منسوخ)', 0, 0, 'منسوخ — از sse_path استفاده کنید', 'text' WHERE NOT EXISTS (SELECT 1 FROM `settings` WHERE `key` = 'wss_path' LIMIT 1); INSERT INTO `settings` (`key`, `value`, `title_fa`, `can_edit`, `is_active`, `description`, `type`) @@ -560,10 +562,13 @@ WHERE `key` = 'node_shared_secret'; UPDATE `settings` SET `title_fa` = COALESCE(NULLIF(TRIM(`title_fa`), ''), 'API سفارش خارجی') WHERE `key` = 'external_order_api_enabled'; -UPDATE `settings` SET `title_fa` = COALESCE(NULLIF(TRIM(`title_fa`), ''), 'قیمت لحظه‌ای از WebSocket') +UPDATE `settings` SET `title_fa` = COALESCE(NULLIF(TRIM(`title_fa`), ''), 'قیمت لحظه‌ای (SSE)') WHERE `key` = 'wss_enabled'; -UPDATE `settings` SET `title_fa` = COALESCE(NULLIF(TRIM(`title_fa`), ''), 'مسیر WebSocket قیمت') +UPDATE `settings` SET `title_fa` = COALESCE(NULLIF(TRIM(`title_fa`), ''), 'مسیر SSE قیمت') +WHERE `key` = 'sse_path'; + +UPDATE `settings` SET `title_fa` = COALESCE(NULLIF(TRIM(`title_fa`), ''), 'مسیر WebSocket (منسوخ)') WHERE `key` = 'wss_path'; UPDATE `settings` SET `title_fa` = COALESCE(NULLIF(TRIM(`title_fa`), ''), 'هاست trigger وب‌سوکت') @@ -672,7 +677,7 @@ UPDATE `settings` SET `type` = 'text' WHERE `key` IN ( 'payping_start_url', 'payping_verify_url', 'payping_gateway_items', 'payping_fin_url', 'payping_fin_token', 'payping_gateway_item', 'tahesab_host', 'tahesab_token', 'tahesab_db', 'node_client_id', 'node_shared_secret', - 'wss_path', 'wss_trigger_host', 'wss_link', 'wss_trigger_path', + 'wss_path', 'sse_path', 'wss_trigger_host', 'wss_link', 'wss_trigger_path', 'tahesab_bank_account', 'telegram_bot_token', 'bale_bot_token', 'support_messenger_id', 'referral_limit', 'referral_bonus_amount', 'balance_report_exclude_customer_id', 'metal_type', 'montify_name', 'price_extreme_date', 'default_vc_number', 'liman_token' @@ -910,7 +915,7 @@ UPDATE `settings` SET `can_edit` = 0 WHERE `key` = 'fee_cash'; -- WebSocket — فقط دیپلوی/SQL (docker / لایت‌اسپید / ws/.env) UPDATE `settings` SET `can_edit` = 0 WHERE `key` IN ( - 'wss_enabled', 'wss_path', 'wss_trigger_host', 'wss_http_port', 'wss_socket_port', + 'wss_enabled', 'sse_path', 'wss_path', 'wss_trigger_host', 'wss_http_port', 'wss_socket_port', 'wss_link', 'wss_trigger_path', 'tahesab_port' ); @@ -1021,8 +1026,9 @@ UPDATE `settings` SET `description` = 'پاسخ JSON آخرین سفارش ار UPDATE `settings` SET `description` = 'شناسه کلاینت برای API سفارش خارجی (هدر X-Client-Id). باید با مقدار تنظیم‌شده در سرویس مقصد یکسان باشد.' WHERE `key` = 'node_client_id'; UPDATE `settings` SET `description` = 'کلید مشترک امضای HMAC برای احراز درخواست‌های API سفارش خارجی. محرمانه نگه دارید.' WHERE `key` = 'node_shared_secret'; UPDATE `settings` SET `description` = 'روشن: API دریافت کیف پول و ثبت سفارش از سرویس فروشگاه خارجی فعال است. نیاز به node_client_id و node_shared_secret دارد.' WHERE `key` = 'external_order_api_enabled'; -UPDATE `settings` SET `description` = 'روشن: اپ کاربر به‌جای polling مکرر /api/price/last و /customer/product/list از WebSocket (wss_path) استفاده می‌کند. خاموش: همیشه polling.' WHERE `key` = 'wss_enabled'; -UPDATE `settings` SET `description` = 'مسیر WebSocket روی دامنه اپ — اپ: wss://{دامنه}{مسیر}. لایت‌اسپید WS Proxy همین مسیر را به wss_socket_port (container:8080) می‌فرستد. پیش‌فرض: /prc/ws' WHERE `key` = 'wss_path'; +UPDATE `settings` SET `description` = 'روشن: اپ به SSE (sse_path) وصل می‌شود و last/list با payload می‌آید؛ خاموش: polling.' WHERE `key` = 'wss_enabled'; +UPDATE `settings` SET `description` = 'مسیر GET لاراول برای استریم قیمت — پیش‌فرض /api/price/stream' WHERE `key` = 'sse_path'; +UPDATE `settings` SET `description` = 'منسوخ — montzar. از sse_path استفاده کنید.', `is_active` = 0 WHERE `key` = 'wss_path'; UPDATE `settings` SET `description` = 'هاست داخلی برای trigger لاراول — معمولاً 127.0.0.1. آدرس نهایی: http://{این_هاست}:{wss_http_port}{wss_trigger_path}' WHERE `key` = 'wss_trigger_host'; UPDATE `settings` SET `description` = 'پورت host برای HTTP سرویس ws (container:3000). لاراول POST برادکست. لایت‌اسپید External App /prc. هر پروژه پورت جدا — مثال طباطبایی: 2012' WHERE `key` = 'wss_http_port'; UPDATE `settings` SET `description` = 'مرجع دیپلوی: پورت host برای WebSocket (container:8080). لایت‌اسپید /prc/ws و WS_SOCKET_PORT در ws/.env. کد PHP/فرانت نمی‌خواند — مثال طباطبایی: 2011' WHERE `key` = 'wss_socket_port'; diff --git a/database/sql/wss_settings_update.sql b/database/sql/wss_settings_update.sql index 09b7002..f65674e 100644 --- a/database/sql/wss_settings_update.sql +++ b/database/sql/wss_settings_update.sql @@ -1,38 +1,22 @@ --- به‌روزرسانی تنظیمات WebSocket قیمت (دیپلوی / پروژه‌های موجود) --- طباطبایی: wss_socket_port=2011، wss_http_port=2012 — هر پروژه جفت پورت خودش +-- مهاجرت WebSocket/montzar → SSE لاراول +-- دیگر نیازی به docker montzar و لایت‌اسپید /prc/ws نیست UPDATE `settings` SET `value` = 'true', `can_edit` = 0, `is_active` = 1, - `description` = 'روشن: اپ از wss_path استفاده می‌کند؛ خاموش: polling /api/price/last و /customer/product/list' + `title_fa` = 'قیمت لحظه‌ای (SSE)', + `description` = 'روشن: اپ به /api/price/stream وصل می‌شود؛ خاموش: polling /api/price/last و /customer/product/list' WHERE `key` = 'wss_enabled'; -UPDATE `settings` SET `value` = '/prc/ws', `can_edit` = 0, `is_active` = 1, - `title_fa` = 'مسیر WebSocket قیمت', - `description` = 'مسیر روی دامنه اپ — wss://{دامنه}{مسیر}. لایت‌اسپید WS Proxy → wss_socket_port (container:8080)' +INSERT INTO `settings` (`key`, `value`, `title_fa`, `can_edit`, `is_active`, `description`, `type`) +SELECT 'sse_path', '/api/price/stream', 'مسیر SSE قیمت', 0, 1, + 'مسیر GET لاراول برای استریم last/list — پیش‌فرض /api/price/stream', + 'text' +WHERE NOT EXISTS (SELECT 1 FROM `settings` WHERE `key` = 'sse_path' LIMIT 1); + +UPDATE `settings` SET `can_edit` = 0, `is_active` = 0, + `title_fa` = 'مسیر WebSocket (منسوخ)', + `description` = 'منسوخ — SSE از sse_path استفاده می‌کند. montzar دیگر لازم نیست.' WHERE `key` = 'wss_path'; -UPDATE `settings` SET `value` = '127.0.0.1', `can_edit` = 0, `is_active` = 1, - `title_fa` = 'هاست trigger وب‌سوکت', - `description` = 'هاست داخلی POST لاراول — http://{این_هاست}:{wss_http_port}{wss_trigger_path}' -WHERE `key` = 'wss_trigger_host'; - -UPDATE `settings` SET `can_edit` = 0, `is_active` = 1, `type` = 'number', - `title_fa` = 'پورت HTTP trigger وب‌سوکت', - `description` = 'پورت host برای container:3000 — لاراول + لایت‌اسپید External App /prc. طباطبایی: 2012' -WHERE `key` = 'wss_http_port'; - -UPDATE `settings` SET `can_edit` = 0, `is_active` = 1, `type` = 'number', - `title_fa` = 'پورت WebSocket (مرجع دیپلوی)', - `description` = 'مرجع docker WS_SOCKET_PORT و لایت‌اسپید /prc/ws (container:8080) — کد نمی‌خواند. طباطبایی: 2011' -WHERE `key` = 'wss_socket_port'; - -UPDATE `settings` SET `value` = '', `can_edit` = 0, `is_active` = 0, - `title_fa` = 'آدرس trigger وب‌سوکت (منسوخ)', - `description` = 'منسوخ — خالی بماند. از wss_trigger_host + wss_http_port استفاده کنید.' -WHERE `key` = 'wss_link'; - -UPDATE `settings` SET `value` = '/prc/api/trigger-update', `can_edit` = 0, `is_active` = 1, - `title_fa` = 'مسیر trigger برادکست', - `description` = 'مسیر POST لاراول — پیش‌فرض /prc/api/trigger-update' -WHERE `key` = 'wss_trigger_path'; - --- پورت‌ها per-project در ws/.env و لایت‌اسپید — مثال طباطبایی: socket=2011، http=2012 +UPDATE `settings` SET `can_edit` = 0, `is_active` = 0, + `description` = 'منسوخ — montzar trigger. با SSE دیگر استفاده نمی‌شود.' +WHERE `key` IN ('wss_trigger_host', 'wss_http_port', 'wss_socket_port', 'wss_trigger_path', 'wss_link'); diff --git a/routes/api.php b/routes/api.php index 2498bca..bece090 100644 --- a/routes/api.php +++ b/routes/api.php @@ -3,6 +3,7 @@ use App\Http\Controllers\Admin\ConstController; use App\Http\Controllers\Site\CityController; +use App\Http\Controllers\Site\PriceStreamController; use App\Http\Controllers\Site\ReportController; use Illuminate\Http\Request; use App\Http\Controllers\Customer\PriceTriggerController; @@ -29,6 +30,7 @@ use App\Http\Controllers\Admin\CategoryController; Route::get('/candle',[ReportController::class, 'candleChart']); Route::get('/price/up',[ReportController::class, 'upDownPrice']); Route::match(['get', 'post'],'/price/last',[ReportController::class, 'lastPrice']); +Route::get('/price/stream', [PriceStreamController::class, 'stream'])->withoutMiddleware(['throttle:api']); Route::get('/price/updatePrise',[ReportController::class, 'updatePrise']); Route::get('/const/key/{key}',[ConstController::class, 'showValue']); Route::get('/check',[ReportController::class, 'check']);