IKAR/lib/backend/ikar_transport.dart

import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:http/http.dart' as http;
import 'package:http/io_client.dart';
import '../core/transport/ikar_network.dart';

/// HTTPS requests and authenticated SignalR events from ikar_server.
class IkarTransport {
  IkarTransport(this.baseUri, {http.Client? client})
    : _client = client ?? IOClient(IkarNetwork.client()) {
    if (baseUri.scheme != 'https' &&
        !['localhost', '127.0.0.1', '::1'].contains(baseUri.host))
      throw ArgumentError('ИКАР требует HTTPS');
  }
  final Uri baseUri;
  final http.Client _client;
  String? token;
  bool interactive = true;
  WebSocket? _socket;
  Timer? _keepAlive;
  final _events = StreamController<Map<String, dynamic>>.broadcast();
  Stream<Map<String, dynamic>> get events => _events.stream;
  bool get eventsConnected => _socket?.readyState == WebSocket.open;
  Future<Map<String, dynamic>> request(
    int opcode,
    Map<dynamic, dynamic> payload,
  ) async {
    final response = await _client
        .post(
          baseUri.resolve('/api/ikar/requests/$opcode'),
          headers: {
            'Content-Type': 'application/json',
            if (token != null) 'Authorization': 'Bearer $token',
            'X-Ikar-Client': 'ikar',
            'X-Ikar-Client-Version': '1.0.4',
          },
          body: jsonEncode({'payload': _json(payload)}),
        )
        .timeout(const Duration(seconds: 30));
    if (response.statusCode == 401)
      return {
        'cmd': 3,
        'opcode': opcode,
        'payload': {'message': 'FAIL_LOGIN_TOKEN'},
      };
    if ([502, 503, 504].contains(response.statusCode)) {
      throw HttpException('Сервер ИКАР недоступен', uri: response.request?.url);
    }
    final result = jsonDecode(response.body);
    if (result is! Map)
      throw const FormatException('Некорректный ответ сервера ИКАР');
    return Map<String, dynamic>.from(result);
  }

  Future<Map<String, dynamic>?> refresh(String refreshToken) async {
    final response = await _client
        .post(
          baseUri.resolve('/api/auth/refresh'),
          headers: {'Content-Type': 'application/json'},
          body: jsonEncode({'refreshToken': refreshToken}),
        )
        .timeout(const Duration(seconds: 30));
    if (response.statusCode != 200) return null;
    return Map<String, dynamic>.from(jsonDecode(response.body) as Map);
  }

  Future<void> connectEvents() async {
    await closeEvents();
    if (token == null) return;
    final uri = baseUri
        .resolve('/hubs/messenger')
        .replace(
          scheme: baseUri.scheme == 'https' ? 'wss' : 'ws',
          port: baseUri.port,
          queryParameters: {
            'access_token': token!,
            'interactive': '$interactive',
          },
        );
    final socket = await IkarNetwork.connectWebSocket(
      uri.toString(),
    ).timeout(const Duration(seconds: 15));
    _socket = socket;
    final handshake = Completer<void>();
    var buffer = '';
    socket.listen(
      (frame) {
        buffer += frame is String ? frame : utf8.decode(frame as List<int>);
        while (buffer.contains('\u001e')) {
          final index = buffer.indexOf('\u001e');
          final record = buffer.substring(0, index);
          buffer = buffer.substring(index + 1);
          if (record.isEmpty) continue;
          final message = jsonDecode(record) as Map;
          if (!handshake.isCompleted && message['type'] == null) {
            if (message['error'] != null) {
              handshake.completeError(StateError('${message['error']}'));
            } else {
              handshake.complete();
            }
          }
          if (message['type'] == 1 && message['target'] == 'QlyraPacket') {
            final args = message['arguments'] as List;
            if (args.isNotEmpty && !_events.isClosed)
              _events.add(Map<String, dynamic>.from(args.first as Map));
          }
          if (message['type'] == 7) unawaited(socket.close());
        }
      },
      onDone: () {
        if (!handshake.isCompleted)
          handshake.completeError(StateError('Соединение событий закрыто'));
        if (identical(_socket, socket)) {
          _socket = null;
          _keepAlive?.cancel();
        }
      },
      onError: (Object error) {
        if (!handshake.isCompleted) handshake.completeError(error);
      },
    );
    socket.add('${jsonEncode({'protocol': 'json', 'version': 1})}\u001e');
    await handshake.future.timeout(const Duration(seconds: 15));
    _keepAlive = Timer.periodic(const Duration(seconds: 10), (_) {
      if (socket.readyState == WebSocket.open) socket.add('{"type":6}\u001e');
    });
  }

  static Object? _json(Object? value) {
    if (value is Map)
      return value.map((key, item) => MapEntry(key.toString(), _json(item)));
    if (value is Iterable) return value.map(_json).toList();
    if (value is BigInt) return value.toInt();
    return value;
  }

  Future<void> closeEvents() async {
    _keepAlive?.cancel();
    final socket = _socket;
    _socket = null;
    await socket?.close();
  }

  Future<void> close() async {
    await closeEvents();
    _client.close();
    await _events.close();
  }
}