minstrel/flutter_client/lib/shared/live_events_provider.dart

// 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<String, dynamic> 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<LiveEvent>((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<LiveEvent>();
  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<void> _runSubscription(Dio dio, StreamController<LiveEvent> controller) async {
  try {
    final resp = await dio.get<ResponseBody>(
      '/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<String, dynamic>;
    return LiveEvent(
      kind: kind ?? (decoded['kind'] as String? ?? ''),
      userId: decoded['user_id'] as String? ?? '',
      data: (decoded['data'] as Map<String, dynamic>?) ?? const {},
    );
  } catch (_) {
    return null;
  }
}