implement powerful viewer count system

2026-05-16 23:38:35 +03:30
parent 95d14f2cec
commit 799581ee52
19 changed files with 762 additions and 17 deletions
--- a/app/Services/Streaming/MediaMtxApiClient.php
+++ b/app/Services/Streaming/MediaMtxApiClient.php
@@ -0,0 +1,60 @@
+<?php
+
+namespace App\Services\Streaming;
+
+use Illuminate\Support\Facades\Http;
+
+class MediaMtxApiClient
+{
+    /**
+     * @return array<int, array{id: string, path: string, query: string, is_cdn: bool}>
+     */
+    public function hlsSessions(): array
+    {
+        $sessions = [];
+        $page = 0;
+
+        do {
+            $payload = (array) Http::baseUrl($this->baseUrl())
+                ->acceptJson()
+                ->connectTimeout((float) config('streaming.mediamtx_api_connect_timeout', 1.0))
+                ->timeout((float) config('streaming.mediamtx_api_timeout', 3.0))
+                ->get('/v3/hlssessions/list', [
+                    'page' => $page,
+                    'itemsPerPage' => max(1, (int) config('streaming.mediamtx_api_items_per_page', 100)),
+                ])
+                ->throw()
+                ->json();
+
+            foreach ((array) ($payload['items'] ?? []) as $session) {
+                if (! is_array($session)) {
+                    continue;
+                }
+
+                $path = trim((string) ($session['path'] ?? ''), '/');
+                $id = (string) ($session['id'] ?? '');
+
+                if ($path === '' || $id === '') {
+                    continue;
+                }
+
+                $sessions[] = [
+                    'id' => $id,
+                    'path' => $path,
+                    'query' => (string) ($session['query'] ?? ''),
+                    'is_cdn' => (bool) ($session['isCDN'] ?? false),
+                ];
+            }
+
+            $page++;
+            $pageCount = max(1, (int) ($payload['pageCount'] ?? $page));
+        } while ($page < $pageCount);
+
+        return $sessions;
+    }
+
+    private function baseUrl(): string
+    {
+        return rtrim((string) config('streaming.mediamtx_api_url'), '/');
+    }
+}
--- a/app/Services/Streaming/MediaMtxService.php
+++ b/app/Services/Streaming/MediaMtxService.php
@@ -10,6 +10,8 @@ use Illuminate\Support\Str;

 class MediaMtxService
 {
+    public function __construct(private readonly ViewerCountStore $viewerCounts) {}
+
    public function authorize(array $payload): bool
    {
        $action = (string) ($payload['action'] ?? '');
@@ -79,6 +81,8 @@ class MediaMtxService
            'live_broadcast_id' => $broadcast->id,
        ])->save();

+        $this->viewerCounts->forget($channel);
+
        return $broadcast;
    }

@@ -107,6 +111,8 @@ class MediaMtxService
            'live_broadcast_id' => null,
        ])->save();

+        $this->viewerCounts->forget($channel);
+
        return $broadcast;
    }

--- a/app/Services/Streaming/MediaMtxViewerCountSynchronizer.php
+++ b/app/Services/Streaming/MediaMtxViewerCountSynchronizer.php
@@ -0,0 +1,100 @@
+<?php
+
+namespace App\Services\Streaming;
+
+use App\Models\Broadcast;
+use App\Models\Channel;
+
+class MediaMtxViewerCountSynchronizer
+{
+    public function __construct(
+        private readonly MediaMtxApiClient $mediaMtx,
+        private readonly ViewerCountStore $viewerCounts,
+        private readonly PlaybackViewerSigner $playbackViewers,
+    ) {}
+
+    /**
+     * @return array{channels: int, viewers: int}
+     */
+    public function sync(): array
+    {
+        $sessionsByPath = $this->sessionsByPath($this->mediaMtx->hlsSessions());
+        $liveChannels = Channel::query()
+            ->where('is_live', true)
+            ->with('liveBroadcast:id,viewer_peak')
+            ->get(['id', 'slug', 'live_broadcast_id', 'viewer_count']);
+
+        $totalViewers = 0;
+
+        foreach ($liveChannels as $channel) {
+            $count = $this->uniqueViewerCount($channel, $sessionsByPath[$channel->slug] ?? []);
+            $totalViewers += $count;
+
+            $this->viewerCounts->put($channel, $count);
+
+            if ($channel->viewer_count !== $count) {
+                $channel->forceFill(['viewer_count' => $count])->save();
+            }
+
+            if ($count > 0 && $channel->live_broadcast_id) {
+                Broadcast::query()
+                    ->whereKey($channel->live_broadcast_id)
+                    ->where('viewer_peak', '<', $count)
+                    ->update(['viewer_peak' => $count]);
+            }
+        }
+
+        Channel::query()
+            ->where('is_live', false)
+            ->where('viewer_count', '>', 0)
+            ->get(['id', 'viewer_count'])
+            ->each(function (Channel $channel): void {
+                $this->viewerCounts->forget($channel);
+                $channel->forceFill(['viewer_count' => 0])->save();
+            });
+
+        return [
+            'channels' => $liveChannels->count(),
+            'viewers' => $totalViewers,
+        ];
+    }
+
+    /**
+     * @param  array<int, array{id: string, path: string, query: string, is_cdn: bool}>  $sessions
+     * @return array<string, array<int, array{id: string, path: string, query: string, is_cdn: bool}>>
+     */
+    private function sessionsByPath(array $sessions): array
+    {
+        $grouped = [];
+
+        foreach ($sessions as $session) {
+            if ($session['is_cdn']) {
+                continue;
+            }
+
+            $grouped[$session['path']][] = $session;
+        }
+
+        return $grouped;
+    }
+
+    /**
+     * @param  array<int, array{id: string, path: string, query: string, is_cdn: bool}>  $sessions
+     */
+    private function uniqueViewerCount(Channel $channel, array $sessions): int
+    {
+        $viewerKeys = [];
+
+        foreach ($sessions as $session) {
+            $viewerKey = $this->playbackViewers->viewerKeyFromQuery(
+                $session['query'],
+                $channel,
+                $channel->liveBroadcast,
+            ) ?? "session:{$session['id']}";
+
+            $viewerKeys[$viewerKey] = true;
+        }
+
+        return count($viewerKeys);
+    }
+}
--- a/app/Services/Streaming/PlaybackViewerSigner.php
+++ b/app/Services/Streaming/PlaybackViewerSigner.php
@@ -0,0 +1,147 @@
+<?php
+
+namespace App\Services\Streaming;
+
+use App\Models\Broadcast;
+use App\Models\Channel;
+use Illuminate\Http\Request;
+use Illuminate\Support\Facades\Cookie;
+use Illuminate\Support\Str;
+use Illuminate\Support\Uri;
+
+class PlaybackViewerSigner
+{
+    private const VIEWER_KEY = 'viewer';
+
+    private const EXPIRES_KEY = 'viewer_expires';
+
+    private const SIGNATURE_KEY = 'viewer_signature';
+
+    public function signedPlaybackUrl(string $hlsUrl, Channel $channel, Broadcast $broadcast, Request $request): string
+    {
+        return (string) Uri::of($hlsUrl)->withQuery(
+            $this->queryParametersForViewerKey(
+                $this->viewerKeyForRequest($request),
+                $channel,
+                $broadcast,
+            ),
+        );
+    }
+
+    /**
+     * @return array{viewer: string, viewer_expires: int, viewer_signature: string}
+     */
+    public function queryParametersForViewerKey(
+        string $viewerKey,
+        Channel $channel,
+        Broadcast $broadcast,
+        ?int $expires = null,
+    ): array {
+        $expires ??= now()->addSeconds($this->signatureTtl())->getTimestamp();
+
+        return [
+            self::VIEWER_KEY => $viewerKey,
+            self::EXPIRES_KEY => $expires,
+            self::SIGNATURE_KEY => $this->signature($viewerKey, $channel->slug, $broadcast->id, $expires),
+        ];
+    }
+
+    public function viewerKeyFromQuery(string $query, Channel $channel, ?Broadcast $broadcast): ?string
+    {
+        if (! $broadcast) {
+            return null;
+        }
+
+        parse_str(ltrim($query, '?'), $parameters);
+
+        $viewerKey = (string) ($parameters[self::VIEWER_KEY] ?? '');
+        $expires = (int) ($parameters[self::EXPIRES_KEY] ?? 0);
+        $signature = (string) ($parameters[self::SIGNATURE_KEY] ?? '');
+
+        if (! $this->isValidViewerKey($viewerKey) || $expires < now()->getTimestamp() || $signature === '') {
+            return null;
+        }
+
+        $expected = $this->signature($viewerKey, $channel->slug, $broadcast->id, $expires);
+
+        return hash_equals($expected, $signature) ? $viewerKey : null;
+    }
+
+    private function viewerKeyForRequest(Request $request): string
+    {
+        if ($request->user()) {
+            return 'u:'.$this->hashIdentity((string) $request->user()->id);
+        }
+
+        return 'g:'.$this->hashIdentity($this->guestVisitorId($request));
+    }
+
+    private function guestVisitorId(Request $request): string
+    {
+        $cookieName = $this->cookieName();
+        $visitorId = (string) $request->cookie($cookieName, '');
+
+        if (! Str::isUuid($visitorId)) {
+            $visitorId = (string) Str::uuid();
+
+            Cookie::queue(Cookie::make(
+                $cookieName,
+                $visitorId,
+                $this->cookieLifetimeMinutes(),
+                (string) config('session.path', '/'),
+                config('session.domain'),
+                config('session.secure'),
+                true,
+                false,
+                config('session.same_site', 'lax'),
+            ));
+        }
+
+        return $visitorId;
+    }
+
+    private function signature(string $viewerKey, string $channelSlug, int $broadcastId, int $expires): string
+    {
+        return hash_hmac(
+            'sha256',
+            "{$channelSlug}|{$broadcastId}|{$viewerKey}|{$expires}",
+            $this->secret(),
+        );
+    }
+
+    private function hashIdentity(string $identity): string
+    {
+        return hash_hmac('sha256', $identity, $this->secret());
+    }
+
+    private function isValidViewerKey(string $viewerKey): bool
+    {
+        return preg_match('/^[ug]:[a-f0-9]{64}$/', $viewerKey) === 1;
+    }
+
+    private function secret(): string
+    {
+        $key = (string) config('app.key');
+
+        if (str_starts_with($key, 'base64:')) {
+            return base64_decode(substr($key, 7), true) ?: $key;
+        }
+
+        return $key;
+    }
+
+    private function cookieName(): string
+    {
+        return (string) config('streaming.viewer_identity_cookie', 'nyone_visitor_id');
+    }
+
+    private function cookieLifetimeMinutes(): int
+    {
+        return max(1, (int) config('streaming.viewer_identity_cookie_lifetime_days', 365)) * 24 * 60;
+    }
+
+    private function signatureTtl(): int
+    {
+        return max(60, (int) config('streaming.viewer_identity_ttl', 86400));
+    }
+}
--- a/app/Services/Streaming/ViewerCountStore.php
+++ b/app/Services/Streaming/ViewerCountStore.php
@@ -0,0 +1,44 @@
+<?php
+
+namespace App\Services\Streaming;
+
+use App\Models\Channel;
+use Illuminate\Contracts\Cache\Repository;
+use Illuminate\Support\Facades\Cache;
+
+class ViewerCountStore
+{
+    public function current(Channel $channel): int
+    {
+        $cached = $this->cache()->get($this->cacheKey($channel));
+
+        return is_numeric($cached) ? max(0, (int) $cached) : max(0, (int) $channel->viewer_count);
+    }
+
+    public function put(Channel $channel, int $count): void
+    {
+        $this->cache()->put($this->cacheKey($channel), max(0, $count), $this->ttl());
+    }
+
+    public function forget(Channel $channel): void
+    {
+        $this->cache()->forget($this->cacheKey($channel));
+    }
+
+    private function cache(): Repository
+    {
+        $store = (string) config('streaming.viewer_count_cache_store', config('cache.default'));
+
+        return $store !== '' ? Cache::store($store) : Cache::store();
+    }
+
+    private function cacheKey(Channel $channel): string
+    {
+        return "streaming:viewers:channels:{$channel->id}";
+    }
+
+    private function ttl(): int
+    {
+        return max(1, (int) config('streaming.viewer_count_ttl', 10));
+    }
+}