SSE Now-Playing Architecture

Status: Live on dev/staging/prod
Last updated: 2026-04-01

Overview

The now-playing feature delivers real-time track metadata to clients via Server-Sent Events (SSE), replacing the previous 3-second HTTP polling approach. This dramatically reduces Cosmos DB load and delivers sub-second updates.

Before vs After

Metric	Polling (old)	SSE (current)
API calls per radio/2h	~2,400	~1–5 (initial fetch + reconnects)
Cosmos DB reads per radio/2h	~2,400	~1–5
Update latency	0–3s	<1s (push on change)
Cosmos 429 errors (now-playing)	~667/2h	~0

---

Architecture

┌──────────────┐     SSE stream      ┌───────────────────────┐
│ Flutter App   │◄────────────────────│  NestJS API           │
│              │     event:           │                       │
│ SSEService   │     now-playing      │  @Sse('now-playing/   │
│              │                      │        events')       │
└──────┬───────┘                      └───────────┬───────────┘
       │                                          │
       │  GET /:id/now-playing                    │  NowPlayingEventService
       │  (initial state only)                    │  (RxJS Subject bus)
       │                                          │
       ▼                                          ▼
┌──────────────┐                      ┌───────────────────────┐
│ REST fallback │                      │  RadiosService        │
│ (one-shot)   │                      │  .getNowPlaying()     │
└──────────────┘                      │                       │
                                      │  Track changed?       │
                                      │  → emit to SSE bus    │
                                      │  → write mem cache    │
                                      │  → persist to MongoDB │
                                      └───────────────────────┘

---

Backend Components

1. SSE Endpoint

File: apps/api/src/app/radios/radios.controller.ts

@Sse('now-playing/events')

• URL: GET /radios/now-playing/events?radioIds=id1,id2
• Query params: radioIds — comma-separated list (optional; omit for all radios)
• Response: text/event-stream with event: now-playing messages
• Auto-reconnect: handled by the browser/client SSE spec

2. Event Bus (NowPlayingEventService)

File: apps/api/src/app/radios/services/now-playing-event.service.ts

• In-memory RxJS Subject<NowPlayingEvent> — no external broker
• emit(dto) — called by the ICY collector (Worker) when a track genuinely changes
• subscribe(radioIds?) — returns Observable<MessageEvent> filtered by radio IDs
• Completes subject on OnModuleDestroy for clean shutdown

3. In-Memory Cache (RadiosService)

File: apps/api/src/app/radios/radios.service.ts

private readonly nowPlayingCache = new Map<string, { data: NowPlayingDto; timestamp: number }>();
private readonly NOW_PLAYING_MEM_CACHE_TTL_MS = 30_000; // 30s

• Fast path: If cached entry is <30s old, return immediately (zero DB queries)
• Write-through: Cache is updated on DB cache hit and on SSE track-change events from the ICY collector
• Purpose: Eliminates redundant MongoDB reads when multiple clients request the same radio within a short window

4. HTTP Cache-Control Header

Cache-Control: public, max-age=15

Applied to GET /:id/now-playing — tells CDNs and browsers to cache the REST response for 15 seconds, further reducing origin hits.

---

Flutter Client Components

NowPlayingSSEService

File: audissey_app/lib/core/services/now_playing_sse_service.dart

• Manages a single SSE connection per active set of radio IDs
• updateRadioIds(Set<String> ids) — dynamically adjusts which radios to subscribe to
• Auto-reconnects on disconnect with exponential backoff
• Exposes Stream<NowPlayingUpdate> for Riverpod providers

Riverpod Providers

File: audissey_app/lib/core/providers/now_playing_provider.dart

Two providers:

1. nowPlayingProvider — Playback-scoped: subscribes to the currently-playing radio, used by the global player bar
   • Its live source is the ICY title read from the player. just_audio keeps the last ICY title across radio switches until the new stream sends its own (Android re-emits it with the new stream's headers), so RadioPlaybackNotifier.isIcyTitleFromPreviousSource scopes ICY to the current playback session: the title held at playRadio, and anything heard while the player is not yet on the radio's playlist item, is dropped until the stream sends a different title. The same rule gates the client report POST /radios/:id/now-playing and the native MediaItem, so a radio without song metadata no longer inherits the previous station's song in the app, nor is it reported to the backend as that radio's (ADO #376). The backend does not filter such reports itself: app versions without the fix can still post them.
2. nowPlayingForRadioProvider — Radio-scoped (StreamProvider.autoDispose.family<NowPlayingUpdate?, String>):
   • Fetches initial state via REST GET /:id/now-playing
   • Subscribes to SSE updates for that radio ID
   • Used by radio_player_screen, home_page, radio_details_screen

Consumer Screens

Screen	Provider	Purpose
radio_player_screen.dart	nowPlayingForRadioProvider	Full now-playing display + music links
home_page.dart	nowPlayingForRadioProvider	Now-playing preview on radio cards
radio_details_screen.dart	nowPlayingForRadioProvider	Now-playing section in radio detail

---

Data Flow (end-to-end)

1. Client opens screen → Riverpod provider fires
2. Initial REST fetch → GET /radios/:id/now-playing → returns cached or fresh data
3. SSE subscription → NowPlayingSSEService.updateRadioIds({id}) opens/reuses connection
4. Track changes on stream → RadiosService.getNowPlaying() detects new StreamTitle
5. Emit to SSE bus → NowPlayingEventService.emit(dto)
6. Push to client → SSE event: now-playing with NowPlayingDto JSON payload
7. UI updates → Riverpod stream updates widget tree reactively

---

Scaling Considerations

• Single-instance limitation: The RxJS Subject is in-memory. If the API scales to multiple replicas, SSE events won't cross instances. Mitigation options:
  • Redis Pub/Sub bridge between instances
  • Sticky sessions (current: single replica per env)
• Connection limits: Each SSE client holds an open HTTP connection. Monitor connection counts on Container Apps if user base grows significantly.
• Memory: The nowPlayingCache Map grows linearly with unique radio IDs requested. At 500 bytes/entry and ~2,000 active radios, this is negligible (1 MB).

---

How to Verify

1. Check SSE stream directly:

   curl -N "https://audissey-api-dev.../api/radios/now-playing/events?radioIds=RADIO_ID"
   

2. Check REST endpoint:

   curl "https://audissey-api-dev.../api/radios/RADIO_ID/now-playing"
   

3. Monitor 429 reduction: Query Log Analytics for Cosmos 429 errors:

   ContainerAppConsoleLogs_CL
   | where Log_s contains "Cosmos 429"
   | summarize count() by bin(TimeGenerated, 1h)