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; + } +}