From 90d8aae51a2c87050b1a12687f5d33f84497a99e Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 12 May 2026 22:25:15 -0400 Subject: [PATCH] feat(#392): Flutter SSE consumer + dispatcher MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Slice 4 — completes the #392 hybrid live-refresh loop. live_events_provider.dart subscribes to /api/events/stream via dio's streaming response mode, parses SSE frames (kind + JSON data + UserID scope), and exposes them as a Riverpod StreamProvider. Heartbeat comments are silently dropped; malformed JSON frames are skipped. The provider auto-rebuilds when auth state changes (token rotation, sign-out → sign-in), so reconnect is implicit. live_events_dispatcher.dart listens to the stream and invalidates the small set of publicly-importable providers we know about: - myQuarantineProvider + homeProvider on any quarantine.* event - homeProvider on any playlist.* event (Home renders the Playlists row) Screen-private providers (library_screen.dart's _liked* / _libraryAlbums / _history, admin screens, etc.) opt in to live-refresh by themselves listening to liveEventsProvider in follow-up commits; the dispatcher stays small and avoids back-edge dependencies on every feature folder. The dispatcher also installs an AppLifecycleState observer for resume-time defensive invalidation. SSE will catch up on its own when the app returns from background, but the invalidate flushes any stale data immediately so the first frame back is fresh. app.dart wires the dispatcher into the post-first-frame callback alongside the other startup activations. Co-Authored-By: Claude Opus 4.7 (1M context) --- flutter_client/lib/app.dart | 6 + .../lib/shared/live_events_dispatcher.dart | 104 +++++++++++++++ .../lib/shared/live_events_provider.dart | 126 ++++++++++++++++++ 3 files changed, 236 insertions(+) create mode 100644 flutter_client/lib/shared/live_events_dispatcher.dart create mode 100644 flutter_client/lib/shared/live_events_provider.dart diff --git a/flutter_client/lib/app.dart b/flutter_client/lib/app.dart index 58073049..78192adf 100644 --- a/flutter_client/lib/app.dart +++ b/flutter_client/lib/app.dart @@ -4,6 +4,7 @@ import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'cache/metadata_prefetcher.dart'; import 'cache/prefetcher.dart'; import 'cache/sync_controller.dart'; +import 'shared/live_events_dispatcher.dart'; import 'shared/routing.dart'; import 'theme/theme_data.dart'; import 'theme/theme_mode_provider.dart'; @@ -33,6 +34,11 @@ class _MinstrelAppState extends ConsumerState { // each section so subsequent taps are drift hits, not network // round trips. ref.read(metadataPrefetcherProvider); + // Live events (#392): subscribes to /api/events/stream and + // invalidates publicly-scoped providers when relevant events + // arrive. Also installs an AppLifecycleState observer for + // resume-time defensive invalidation. + ref.read(liveEventsDispatcherProvider); }); } diff --git a/flutter_client/lib/shared/live_events_dispatcher.dart b/flutter_client/lib/shared/live_events_dispatcher.dart new file mode 100644 index 00000000..28bc65f0 --- /dev/null +++ b/flutter_client/lib/shared/live_events_dispatcher.dart @@ -0,0 +1,104 @@ +// Maps incoming LiveEvent kinds to provider invalidations. +// +// Activated by ref.read(liveEventsDispatcherProvider) in app.dart; once +// activated, the dispatcher listens to liveEventsProvider for the +// lifetime of the ProviderScope and invalidates the small set of +// public-scoped providers we know about. +// +// Screen-scoped providers (file-private providers in library_screen.dart, +// admin handlers, etc.) opt in to live-refresh by themselves listening +// to liveEventsProvider — the dispatcher only handles cross-screen, +// publicly-importable providers. This keeps the dispatcher small and +// avoids a back-edge dependency from /shared onto every feature folder. +// +// Includes an AppLifecycleState resume handler that defensively +// invalidates the same set on app foreground. SSE will catch up on its +// own, but on cold-start or after a long background period the +// re-invalidate is fast and avoids stale data flashing before the +// stream reconnects. + +import 'package:flutter/widgets.dart'; +import 'package:flutter_riverpod/flutter_riverpod.dart'; + +import '../library/library_providers.dart' show homeProvider; +import '../quarantine/quarantine_provider.dart'; +import 'live_events_provider.dart'; + +/// Activates the SSE → invalidation dispatcher for the lifetime of the +/// containing ProviderScope. Read this once at startup (from app.dart); +/// the provider's body runs once, sets up listeners, and returns a +/// sentinel value. +final liveEventsDispatcherProvider = Provider<_LiveEventsDispatcher>((ref) { + final d = _LiveEventsDispatcher(ref); + ref.onDispose(d.dispose); + return d; +}); + +class _LiveEventsDispatcher with WidgetsBindingObserver { + _LiveEventsDispatcher(this._ref) { + // Subscribe to the event stream. Each emitted event invokes + // _handle. Errors / disconnects auto-rebuild the underlying + // provider; we don't need to react to them here. + _sub = _ref.listen>( + liveEventsProvider, + (_, next) => next.whenData(_handle), + fireImmediately: false, + ); + WidgetsBinding.instance.addObserver(this); + } + + final Ref _ref; + late final ProviderSubscription> _sub; + + void _handle(LiveEvent e) { + // Map event kinds to provider invalidations. Keep this list short: + // only providers reachable from /shared. Screen-private providers + // listen to liveEventsProvider themselves. + switch (e.kind) { + case 'quarantine.flagged': + case 'quarantine.unflagged': + case 'quarantine.resolved': + case 'quarantine.file_deleted': + case 'quarantine.deleted_via_lidarr': + _ref.invalidate(myQuarantineProvider); + // Hidden / liked / album rows referencing a deleted track may + // now be stale — homeProvider re-fetch refreshes the cards. + _ref.invalidate(homeProvider); + case 'playlist.created': + case 'playlist.updated': + case 'playlist.deleted': + case 'playlist.tracks_changed': + // Home renders a Playlists row; refresh it so a delete or add + // is reflected immediately. Detail screens that need the + // mutated row will get a separate invalidate from their own + // listener (filed as a follow-up). + _ref.invalidate(homeProvider); + case 'scan.run_started': + case 'scan.run_finished': + // Admin scan card provider lives in the admin feature folder + // and is file-private today; will be invalidated by the admin + // dashboard's own listener once that exists. + break; + default: + // track.liked / track.unliked / request.status_changed reach + // screen-private providers; their screens listen directly. + break; + } + } + + @override + void didChangeAppLifecycleState(AppLifecycleState state) { + if (state == AppLifecycleState.resumed) { + // Defensive cold-start invalidation. SSE will catch up on its + // own, but the re-invalidate flushes any stale data that + // landed while the app was backgrounded. + _ref.invalidate(myQuarantineProvider); + _ref.invalidate(homeProvider); + } + } + + void dispose() { + WidgetsBinding.instance.removeObserver(this); + _sub.close(); + } +} diff --git a/flutter_client/lib/shared/live_events_provider.dart b/flutter_client/lib/shared/live_events_provider.dart new file mode 100644 index 00000000..22b840bd --- /dev/null +++ b/flutter_client/lib/shared/live_events_provider.dart @@ -0,0 +1,126 @@ +// Subscribes to the server's Server-Sent Events stream +// (GET /api/events/stream, see Fable #392) and exposes a parsed event +// stream as a Riverpod StreamProvider. Consumers wire invalidation +// behavior in live_events_dispatcher.dart. +// +// On disconnect (network blip, server restart, token rotation), the +// provider's StreamController completes; the dispatcher's auto-rebuild +// when the auth state changes re-subscribes. Explicit exponential +// backoff lives at the dispatcher layer so we don't re-create the dio +// connection too aggressively. + +import 'dart:async'; +import 'dart:convert'; + +import 'package:dio/dio.dart'; +import 'package:flutter_riverpod/flutter_riverpod.dart'; + +import '../auth/auth_provider.dart'; +import '../library/library_providers.dart' show dioProvider; + +/// Parsed event from the server's SSE stream. `kind` follows +/// "domain.action" naming; `data` carries the payload map. +class LiveEvent { + const LiveEvent({required this.kind, required this.userId, required this.data}); + + final String kind; + final String userId; + final Map data; + + @override + String toString() => 'LiveEvent($kind, user=$userId, data=$data)'; +} + +/// Streams parsed LiveEvent values from /api/events/stream. Errors and +/// disconnects surface as stream errors; the dispatcher decides whether +/// to retry. +final liveEventsProvider = StreamProvider((ref) async* { + // Gate the subscription on having both a server URL and a session + // token. If either is missing, emit nothing and let the provider + // auto-rebuild when auth state lands. + final token = await ref.watch(sessionTokenProvider.future); + if (token == null || token.isEmpty) { + return; + } + final dio = await ref.watch(dioProvider.future); + + final controller = StreamController(); + ref.onDispose(controller.close); + + // ignore: unawaited_futures + _runSubscription(dio, controller); + + yield* controller.stream; +}); + +/// Runs the dio streaming request and pushes parsed events into +/// [controller]. Closes the controller when the stream ends or errors. +Future _runSubscription(Dio dio, StreamController controller) async { + try { + final resp = await dio.get( + '/api/events/stream', + options: Options( + responseType: ResponseType.stream, + // No timeout — the server emits 15s heartbeats; idle timeouts + // on the client side would tear down a healthy connection. + receiveTimeout: Duration.zero, + headers: const {'Accept': 'text/event-stream'}, + ), + ); + + // SSE frames are delimited by blank lines. Accumulate raw bytes + // into a string buffer; flush parsed events on each "\n\n". + var buffer = ''; + await for (final chunk in resp.data!.stream) { + buffer += utf8.decode(chunk, allowMalformed: true); + while (true) { + final i = buffer.indexOf('\n\n'); + if (i < 0) break; + final frame = buffer.substring(0, i); + buffer = buffer.substring(i + 2); + final event = _parseFrame(frame); + if (event != null && !controller.isClosed) { + controller.add(event); + } + } + } + if (!controller.isClosed) { + await controller.close(); + } + } catch (e, st) { + if (!controller.isClosed) { + controller.addError(e, st); + await controller.close(); + } + } +} + +/// Parses one SSE frame. Heartbeat comments (starting with ":") and +/// frames without a `data:` line return null. Frames with a `data:` +/// payload that doesn't parse as JSON are also dropped (logged at +/// debug level by the caller if needed). +LiveEvent? _parseFrame(String frame) { + String? kind; + String? dataLine; + for (final line in frame.split('\n')) { + if (line.isEmpty || line.startsWith(':')) { + continue; + } + if (line.startsWith('event:')) { + kind = line.substring(6).trim(); + } else if (line.startsWith('data:')) { + dataLine = line.substring(5).trim(); + } + } + if (dataLine == null) return null; + try { + final decoded = jsonDecode(dataLine) as Map; + return LiveEvent( + kind: kind ?? (decoded['kind'] as String? ?? ''), + userId: decoded['user_id'] as String? ?? '', + data: (decoded['data'] as Map?) ?? const {}, + ); + } catch (_) { + return null; + } +}