boursecu/boursecu_server
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added assets into stream

Browse Source
This commit is contained in:
mohmdsdg
2026-06-27 15:51:44 +03:30
parent a0b3d8d3f5
commit 7b05db0ff0
6 changed files with 170 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/Services/PriceStreamService.php
+83 -14
View File
@@ -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<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);
}
/**
@@ -98,8 +113,57 @@ class PriceStreamService
return;
}
$key = $this->inboxKey();
$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'] : [];
@@ -126,11 +190,6 @@ class PriceStreamService
});
}
private function inboxKey(): string
{
return self::INBOX_PREFIX.'global';
}
/**
* @param array<string, mixed>|array<int, mixed> $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));
}