Update Centrifugo to send initial NP/time payload.

config/routes/api.php

@@ -69,6 +69,9 @@ return static function (RouteCollectorProxy $app) {
                     $group->post('/sftp-event', Controller\Api\Internal\SftpEventAction::class)
                         ->setName('api:internal:sftp-event');
 
+                    $group->post('/centrifugo', Controller\Api\Internal\CentrifugoAction::class)
+                        ->setName('api:internal:centrifugo');
+
                     $group->get('/relays', Controller\Api\Internal\RelaysController::class)
                         ->setName('api:internal:relays')
                         ->add(Middleware\RequireLogin::class);

frontend/src/functions/useNowPlaying.ts

@@ -25,6 +25,7 @@ export const nowPlayingProps = {
 
 export default function useNowPlaying(props) {
     const np: ShallowRef<ApiNowPlaying> = shallowRef(NowPlaying);
+    const npTimestamp: Ref<number> = ref(0);
 
     const currentTime: Ref<number> = ref(Math.floor(Date.now() / 1000));
     const currentTrackDuration: Ref<number> = ref(0);
@@ -32,6 +33,7 @@ export default function useNowPlaying(props) {
 
     const setNowPlaying = (np_new: ApiNowPlaying) => {
         np.value = np_new;
+        npTimestamp.value = Date.now();
 
         currentTrackDuration.value = np_new.now_playing.duration ?? 0;
 
@@ -51,34 +53,7 @@ export default function useNowPlaying(props) {
         }));
     }
 
-    const nowPlayingUri = props.useStatic
-        ? getApiUrl(`/nowplaying_static/${props.stationShortName}.json`)
-        : getApiUrl(`/nowplaying/${props.stationShortName}`);
-
-    const timeUri = getApiUrl('/time');
-    const {axiosSilent} = useAxios();
-
-    const axiosNoCacheConfig = {
-        headers: {
-            'Cache-Control': 'no-cache',
-            'Pragma': 'no-cache',
-            'Expires': '0',
-        }
-    };
-
     if (props.useSse) {
-        // Make an initial AJAX request before SSE takes over.
-        onMounted(() => {
-            axiosSilent.get(nowPlayingUri.value, axiosNoCacheConfig).then((response) => {
-                setNowPlaying(response.data);
-            });
-
-            axiosSilent.get(timeUri.value, axiosNoCacheConfig).then((response) => {
-                currentTime.value = response.data.timestamp;
-            });
-        });
-
-        // Subsequent events come from SSE.
         const sseBaseUri = getApiUrl('/live/nowplaying/sse');
         const sseUriParams = new URLSearchParams({
             "cf_connect": JSON.stringify({
@@ -90,23 +65,50 @@ export default function useNowPlaying(props) {
         });
         const sseUri = sseBaseUri.value + '?' + sseUriParams.toString();
 
+        const handleSseData = (ssePayload) => {
+            const jsonData = ssePayload?.pub?.data ?? {};
+            if (ssePayload.channel === 'global:time') {
+                currentTime.value = jsonData.time;
+            } else {
+                if (npTimestamp.value === 0) {
+                    setNowPlaying(jsonData.np);
+                } else {
+                    // SSE events often dispatch *too quickly* relative to the delays involved in
+                    // Liquidsoap and Icecast, so we delay these changes from showing up to better
+                    // approximate when listeners will really hear the track change.
+                    setTimeout(() => {
+                        setNowPlaying(jsonData.np);
+                    }, 3000);
+                }
+            }
+        }
+
         const {data} = useEventSource(sseUri);
         watch(data, (dataRaw: string) => {
             const jsonData: SSEResponse = JSON.parse(dataRaw);
-            const jsonDataNp = jsonData?.pub?.data ?? {};
-
-            if ('np' in jsonDataNp) {
-                // SSE events often dispatch *too quickly* relative to the delays involved in
-                // Liquidsoap and Icecast, so we delay these changes from showing up to better
-                // approximate when listeners will really hear the track change.
-                setTimeout(() => {
-                    setNowPlaying(jsonDataNp.np);
-                }, 3000);
-            } else if ('time' in jsonDataNp) {
-                currentTime.value = jsonDataNp.time;
+            if ('connect' in jsonData) {
+                const initialData = jsonData.connect.data ?? [];
+                initialData.forEach((initialRow) => handleSseData(initialRow));
+            } else if ('channel' in jsonData) {
+                handleSseData(jsonData);
             }
         });
     } else {
+        const nowPlayingUri = props.useStatic
+            ? getApiUrl(`/nowplaying_static/${props.stationShortName}.json`)
+            : getApiUrl(`/nowplaying/${props.stationShortName}`);
+
+        const timeUri = getApiUrl('/time');
+        const {axiosSilent} = useAxios();
+
+        const axiosNoCacheConfig = {
+            headers: {
+                'Cache-Control': 'no-cache',
+                'Pragma': 'no-cache',
+                'Expires': '0',
+            }
+        };
+
         const checkNowPlaying = () => {
             axiosSilent.get(nowPlayingUri.value, axiosNoCacheConfig).then((response) => {
                 setNowPlaying(response.data);

src/Controller/Api/Internal/CentrifugoAction.php

@@ -0,0 +1,67 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Controller\Api\Internal;
+
+use App\Cache\NowPlayingCache;
+use App\Container\LoggerAwareTrait;
+use App\Controller\SingleActionInterface;
+use App\Http\Response;
+use App\Http\ServerRequest;
+use App\Service\Centrifugo;
+use App\Utilities\Types;
+use Psr\Http\Message\ResponseInterface;
+
+final class CentrifugoAction implements SingleActionInterface
+{
+    use LoggerAwareTrait;
+
+    public function __construct(
+        private readonly NowPlayingCache $npCache
+    ) {
+    }
+
+    public function __invoke(ServerRequest $request, Response $response, array $params): ResponseInterface
+    {
+        $parsedBody = Types::array($request->getParsedBody());
+        $this->logger->debug('Centrifugo connection body', $parsedBody);
+
+        $channels = array_filter(
+            $parsedBody['channels'] ?? [],
+            fn($channel) => str_starts_with($channel, 'station:') || $channel === Centrifugo::GLOBAL_TIME_CHANNEL
+        );
+
+        $allInitialData = [];
+
+        foreach ($channels as $channel) {
+            $initialData = [];
+
+            if ($channel === Centrifugo::GLOBAL_TIME_CHANNEL) {
+                $initialData['time'] = time();
+            } elseif (str_starts_with($channel, 'station:')) {
+                $stationName = substr($channel, 8);
+                $np = $this->npCache->getForStation($stationName);
+                if (null === $np) {
+                    continue;
+                }
+                $initialData['np'] = $np;
+            }
+
+            $allInitialData[] = [
+                'channel' => $channel,
+                'pub' => [
+                    'data' => $initialData,
+                ],
+            ];
+        }
+
+        return $response->withJson([
+            'result' => [
+                'user' => '',
+                'channels' => $channels,
+                'data' => $allInitialData,
+            ],
+        ]);
+    }
+}

util/docker/web/centrifugo/config.yaml.tmpl

@@ -8,6 +8,8 @@ websocket_disable: true
 uni_websocket: true
 uni_sse: true
 uni_http_stream: true
+proxy_connect_endpoint: http://127.0.0.1:6010/api/internal/centrifugo
+proxy_connect_timeout: 10s
 allowed_origins:
   - "*"