petitspas/frontend/lib/services/api/cards_stream_connector_io.dart

import 'dart:async';
import 'dart:convert';

import 'package:http/http.dart' as http;

/// Connecteur SSE via HTTP stream (IO / mobile) — ticket #195.
class CardsStreamConnector {
  http.Client? _client;
  StreamSubscription<List<int>>? _sub;
  bool _closed = false;

  Future<void> connect({
    required String url,
    String? bearerToken,
    required void Function(String eventType, String data) onEvent,
    required void Function(Object error) onError,
  }) async {
    close();
    _closed = false;
    final client = http.Client();
    _client = client;

    try {
      final request = http.Request('GET', Uri.parse(url));
      request.headers['Accept'] = 'text/event-stream';
      request.headers['Cache-Control'] = 'no-cache';
      if (bearerToken != null && bearerToken.isNotEmpty) {
        request.headers['Authorization'] = 'Bearer $bearerToken';
      }

      final response = await client.send(request);
      if (response.statusCode != 200) {
        onError(
          StateError('SSE cards HTTP ${response.statusCode}'),
        );
        close();
        return;
      }

      String? eventType;
      final dataBuf = StringBuffer();
      final lineBuf = StringBuffer();

      void flushEvent() {
        if (dataBuf.isEmpty) {
          eventType = null;
          return;
        }
        onEvent(eventType ?? 'message', dataBuf.toString());
        dataBuf.clear();
        eventType = null;
      }

      void handleLine(String line) {
        if (line.isEmpty) {
          flushEvent();
          return;
        }
        if (line.startsWith(':')) return;
        if (line.startsWith('event:')) {
          eventType = line.substring(6).trim();
          return;
        }
        if (line.startsWith('data:')) {
          final payload = line.length > 5 ? line.substring(5) : '';
          final trimmed = payload.startsWith(' ')
              ? payload.substring(1)
              : payload;
          if (dataBuf.isNotEmpty) dataBuf.write('\n');
          dataBuf.write(trimmed);
        }
      }

      _sub = response.stream.listen(
        (chunk) {
          if (_closed) return;
          lineBuf.write(utf8.decode(chunk, allowMalformed: true));
          var content = lineBuf.toString();
          var idx = content.indexOf('\n');
          while (idx >= 0) {
            var line = content.substring(0, idx);
            if (line.endsWith('\r')) {
              line = line.substring(0, line.length - 1);
            }
            handleLine(line);
            content = content.substring(idx + 1);
            idx = content.indexOf('\n');
          }
          lineBuf
            ..clear()
            ..write(content);
        },
        onError: (Object e) {
          if (!_closed) onError(e);
        },
        onDone: () {
          if (!_closed) onError(StateError('SSE cards fermé'));
        },
        cancelOnError: true,
      );
    } catch (e) {
      if (!_closed) onError(e);
      close();
    }
  }

  void close() {
    _closed = true;
    _sub?.cancel();
    _sub = null;
    _client?.close();
    _client = null;
  }
}