From 7b05db0ff03e617453fba8c37683961fee6e7d70 Mon Sep 17 00:00:00 2001 From: adonis Date: Sat, 27 Jun 2026 15:51:44 +0330 Subject: [PATCH] added assets into stream --- .../Controllers/Admin/WalletController.php | 18 ++++ .../Single/SingleOrderController.php | 9 ++ .../Site/PriceStreamController.php | 26 ++++- app/Services/PriceStreamService.php | 97 ++++++++++++++++--- app/Services/WalletBroadcastService.php | 27 ++++++ routes/api.php | 9 +- 6 files changed, 170 insertions(+), 16 deletions(-) create mode 100644 app/Services/WalletBroadcastService.php diff --git a/app/Http/Controllers/Admin/WalletController.php b/app/Http/Controllers/Admin/WalletController.php index 5d5aceb..a642465 100644 --- a/app/Http/Controllers/Admin/WalletController.php +++ b/app/Http/Controllers/Admin/WalletController.php @@ -24,6 +24,7 @@ use App\Models\TransactionsAdmin; use App\Services\BalanceReportService; use App\Services\ConfigService; use App\Services\WalletBalanceService; +use App\Services\WalletBroadcastService; use Hekmatinasser\Verta\Verta; @@ -154,6 +155,8 @@ class WalletController extends Controller . "\n" . $this->Config('name_fa'); $this->Send($customer->mobile, $message); + $this->broadcastCustomerWalletAssets($customer); + return response()->json(true); } public function charge_gold(Request $request) @@ -224,6 +227,8 @@ class WalletController extends Controller . $data['note'] . $this->Config('name_fa'); $this->Send($customer->mobile, $message); + $this->broadcastCustomerWalletAssets($customer); + return response()->json(true); } @@ -367,6 +372,7 @@ class WalletController extends Controller . "\n" . $this->Config('name_fa'); $this->Send($customer->mobile, $message); + $this->broadcastCustomerWalletAssets($customer); return response()->json(true); } @@ -467,6 +473,10 @@ class WalletController extends Controller $transaction->save(); $message = 'تراکنش با موفقیت به‌روزرسانی شد.'; + $customer = Customer::find($transaction->customer_id); + if ($customer) { + $this->broadcastCustomerWalletAssets($customer); + } return response()->json([ 'message' => $message, @@ -660,4 +670,12 @@ class WalletController extends Controller ]); } + private function broadcastCustomerWalletAssets(Customer $customer): void + { + app(WalletBroadcastService::class)->broadcastForCustomer( + (int) $customer->id, + (int) $customer->type + ); + } + } diff --git a/app/Http/Controllers/Single/SingleOrderController.php b/app/Http/Controllers/Single/SingleOrderController.php index bbe51ec..d4a36f0 100644 --- a/app/Http/Controllers/Single/SingleOrderController.php +++ b/app/Http/Controllers/Single/SingleOrderController.php @@ -25,6 +25,7 @@ use App\Models\Setting; use App\Models\Cnt; use App\Services\MontzarService; use App\Services\ConfigService; +use App\Services\WalletBroadcastService; class SingleOrderController extends Controller @@ -221,6 +222,14 @@ class SingleOrderController extends Controller report($e); } } + + if ($customer) { + app(WalletBroadcastService::class)->broadcastForCustomer( + (int) $customer->id, + $ownerType + ); + } + return response()->json([$response, $data]); } diff --git a/app/Http/Controllers/Site/PriceStreamController.php b/app/Http/Controllers/Site/PriceStreamController.php index c4afcd6..bc2584d 100644 --- a/app/Http/Controllers/Site/PriceStreamController.php +++ b/app/Http/Controllers/Site/PriceStreamController.php @@ -3,6 +3,7 @@ namespace App\Http\Controllers\Site; use App\Http\Controllers\Controller; +use App\Services\AssetSummaryService; use App\Services\LivePriceCacheService; use App\Services\PriceBroadcastService; use App\Services\PriceStreamService; @@ -19,7 +20,11 @@ class PriceStreamController extends Controller abort(404); } - return response()->stream(function () use ($service) { + $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')) { @@ -50,6 +55,18 @@ class PriceStreamController extends Controller $send('list', $list); } + $customerLastId = 0; + if ($customerId > 0) { + $assets = app(AssetSummaryService::class)->buildForCustomer( + $customerId, + $customerOwnerType > 0 ? $customerOwnerType : 2 + ); + if ($assets !== []) { + $send('assets', $assets); + } + $customerLastId = $service->currentCustomerSeq($customerId); + } + $lastId = $service->currentSeq(); $heartbeatTicks = 0; @@ -62,6 +79,13 @@ class PriceStreamController extends Controller $lastId = (int) $event['id']; } + if ($customerId > 0) { + foreach ($service->consumeCustomerSince($customerId, $customerLastId) as $event) { + $send((string) $event['event'], $event['data']); + $customerLastId = (int) $event['id']; + } + } + $heartbeatTicks++; // هر ۵ ثانیه — LiteSpeed/پروکسی اتصال بیکار را نمی‌بندد if ($heartbeatTicks >= 10) { diff --git a/app/Services/PriceStreamService.php b/app/Services/PriceStreamService.php index 12a7b8c..eaab4d0 100644 --- a/app/Services/PriceStreamService.php +++ b/app/Services/PriceStreamService.php @@ -9,6 +9,8 @@ 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; @@ -70,9 +72,16 @@ class PriceStreamService public function currentSeq(): int { - $inbox = Cache::get($this->inboxKey(), ['seq' => 0, 'events' => []]); + return $this->currentInboxSeq($this->inboxKey()); + } - return (int) ($inbox['seq'] ?? 0); + public function currentCustomerSeq(int $customerId): int + { + if ($customerId <= 0) { + return 0; + } + + return $this->currentInboxSeq($this->customerInboxKey($customerId)); } /** @@ -80,13 +89,19 @@ class PriceStreamService */ public function consumeSince(int $afterId): array { - $inbox = Cache::get($this->inboxKey(), ['seq' => 0, 'events' => []]); - $events = is_array($inbox['events'] ?? null) ? $inbox['events'] : []; + return $this->consumeInboxSince($this->inboxKey(), $afterId); + } - return array_values(array_filter( - $events, - static fn (array $event): bool => (int) ($event['id'] ?? 0) > $afterId - )); + /** + * @return array + */ + public function consumeCustomerSince(int $customerId, int $afterId): array + { + if ($customerId <= 0) { + return []; + } + + return $this->consumeInboxSince($this->customerInboxKey($customerId), $afterId); } /** @@ -98,8 +113,57 @@ class PriceStreamService return; } - $key = $this->inboxKey(); + $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); + } + + 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 + */ + 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|array $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'] : []; @@ -126,11 +190,6 @@ class PriceStreamService }); } - private function inboxKey(): string - { - return self::INBOX_PREFIX.'global'; - } - /** * @param array|array $data */ @@ -148,6 +207,16 @@ class PriceStreamService 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)); } diff --git a/app/Services/WalletBroadcastService.php b/app/Services/WalletBroadcastService.php new file mode 100644 index 0000000..0ad8891 --- /dev/null +++ b/app/Services/WalletBroadcastService.php @@ -0,0 +1,27 @@ +isEnabled() || $customerId <= 0) { + return; + } + + if ($ownerType === null || $ownerType <= 0) { + $ownerType = (int) Customer::query()->whereKey($customerId)->value('type'); + } + + if ($ownerType <= 0) { + $ownerType = 2; + } + + $payload = app(AssetSummaryService::class)->buildForCustomer($customerId, $ownerType); + $stream->publishForCustomer($customerId, 'assets', $payload); + } +} diff --git a/routes/api.php b/routes/api.php index bece090..45d9f0d 100644 --- a/routes/api.php +++ b/routes/api.php @@ -30,7 +30,14 @@ 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']); + +$priceStreamRoute = Route::get('/price/stream', [PriceStreamController::class, 'stream']) + ->withoutMiddleware(['throttle:api']); + +if (! app(ConfigService::class)->isStore()) { + $priceStreamRoute->middleware('auth:customer'); +} + Route::get('/price/updatePrise',[ReportController::class, 'updatePrise']); Route::get('/const/key/{key}',[ConstController::class, 'showValue']); Route::get('/check',[ReportController::class, 'check']);