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