eatvio_app/lib/core/websockets/ws_manager.dart

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

import 'package:dart_pusher_channels/dart_pusher_channels.dart';
import 'package:easy_debounce/easy_debounce.dart';
import 'package:eatvio_app/core/app_config.dart';
import 'package:eatvio_app/core/const.dart';
import 'package:eatvio_app/core/locator_service.dart';
import 'package:eatvio_app/core/websockets/ws_service.dart';
import 'package:eatvio_app/core/token_storage.dart';
import 'package:eatvio_app/core/error_reporting/error_reporter.dart';
import 'package:eatvio_app/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart';
import 'package:eatvio_app/features/notifications/presentation/notifications_cubit/notifications_cubit.dart';

class WsManager {
  final WsService wsService;
  final ErrorReporter _errorReporter;

  WsManager({required this.wsService, required ErrorReporter errorReporter})
      : _errorReporter = errorReporter;

  PusherChannelsClient? client;
  int? userId;

  PrivateChannel? _privateNotificationChannel;
  PrivateChannel? _privateMessagesChannel;
  PrivateChannel? _privateDirectChatChannel;
  PresenceChannel? _presenceDirectChatChannel;

  StreamSubscription? _presenceMemberAddedSubscription;
  StreamSubscription? _presenceMemberRemovedSubscription;
  StreamSubscription? _presenceSubscriptionSucceededSubscription;
  final List<_PresenceDirectChatListener> _presenceDirectChatListeners = [];
  bool _presenceChannelEventsBound = false;

  StreamSubscription? _lifecycleSubscription;
  StreamSubscription? _eventSubscription;
  StreamSubscription? _connectionEstablishedSubscription;

  Stream<ChannelReadEvent>? get newMessagesStream => _privateMessagesChannel?.bind('App\\Events\\ChatMessageUpdated');

  Stream<ChannelReadEvent>? get newNotificationsStream => _privateNotificationChannel?.bind('Illuminate\\Notifications\\Events\\BroadcastNotificationCreated');

  Stream<ChannelReadEvent>? get directChatStream => _privateDirectChatChannel?.bind('App\\Events\\ChatMessageUpdated');

  Stream<ChannelReadEvent>? get clientTypingStream => _privateDirectChatChannel?.bind('client-typing');

  Future<void> initialize() async {
    if (client == null) {
      try {
        if (!await TokenStorage.haveTokenInStorage()) return;
        final wsUrl = await wsService.getWebsocket();

        final options = PusherChannelsOptions.fromHost(
          scheme: 'wss',
          key: wsUrl.key,
          host: wsUrl.host,
          shouldSupplyMetadataQueries: true,
          metadata: const PusherChannelsOptionsMetadata.byDefault(),
          port: 443,
        );

        client = PusherChannelsClient.websocket(
          options: options,
          connectionErrorHandler: (exception, trace, refresh) {
            refresh();
          },
          minimumReconnectDelayDuration: const Duration(
            seconds: 1,
          ),
          defaultActivityDuration: const Duration(
            seconds: 30,
          ),
          waitForPongDuration: const Duration(
            seconds: 30,
          ),
        );

        String? extractSocketId(String? data) {
          if (data == null) return null;
          try {
            final decoded = jsonDecode(data);
            return decoded['socket_id'] as String?;
          } catch (_) {
            return null;
          }
        }

        _connectionEstablishedSubscription = client?.eventStream.listen((event) async {
          if (event.name == 'pusher:connection_established') {
            final data = event.data;
            final socketId = extractSocketId(data);
            await SocketIdStorage.saveSocketIdInStorage(socketId!);
          }
        });

        await client?.connect();

        client = client;
        userId = wsUrl.userId;

        await _initPrivateNotificationsChannel();
        await _initPrivateMessagesChannel();

        _lifecycleSubscription = client?.lifecycleStream.listen((data) {
          if (data == PusherChannelsClientLifeCycleState.establishedConnection) {
            _unsubscribe();
            _subscribe();
          }
        });
        _eventSubscription = client?.eventStream.listen((event) {
          log('WsManager received event: ${event.name}');
          // Только если это не системные сообщения Pusher, обновляем список чатов
          if (!event.name.startsWith('pusher:')) {
            EasyDebounce.debounce(
              'searchChatRoom',
              const Duration(milliseconds: 300),
              () => sl<ChatsCubit>().getChatRooms(),
            );
          }
        });
        _subscribe();
        await _connectCubitsToStreams();
      } catch (e, st) {
        await _errorReporter.report(e, st, origin: 'WsManager');
        log(e.toString(), stackTrace: st);
      }
    }
  }

  /// Подключает ChatsCubit и NotificationsCubit к стримам сокета.
  /// Вызывается после успешного initialize(), чтобы уведомления и чаты работали
  /// и при входе уже залогиненного пользователя (редирект на feed), не только после экрана логина.
  Future<void> _connectCubitsToStreams() async {
    try {
      await sl<ChatsCubit>().connectToWebSocket();
      await sl<NotificationsCubit>().getNotifications();
      await sl<NotificationsCubit>().connectToWebSocket();
    } catch (e, st) {
      log('WsManager._connectCubitsToStreams: $e', stackTrace: st);
      await _errorReporter.report(e, st, origin: 'WsManager');
    }
  }

  Future<void> _initPrivateNotificationsChannel() async {
    final token = await TokenStorage.getTokenFromStorage();
    _privateNotificationChannel = client?.privateChannel(
      '$newNotificationsChannelName.$userId',
      authorizationDelegate: EndpointAuthorizableChannelTokenAuthorizationDelegate.forPrivateChannel(
        authorizationEndpoint: Uri.parse('${sl<AppConfig>().baseUrl}/broadcasting/auth'),
        headers: {
          'Authorization': 'Bearer $token',
        },
      ),
    );
  }

  Future<void> _initPrivateMessagesChannel() async {
    final token = await TokenStorage.getTokenFromStorage();
    _privateMessagesChannel = client?.privateChannel(
      '$newMessagesChannelName.$userId',
      authorizationDelegate: EndpointAuthorizableChannelTokenAuthorizationDelegate.forPrivateChannel(
        authorizationEndpoint: Uri.parse('${sl<AppConfig>().baseUrl}/broadcasting/auth'),
        headers: {
          'Authorization': 'Bearer $token',
        },
      ),
    );
  }

  Future<void> initPrivateDirectChatChannel(String chatId) async {
    final token = await TokenStorage.getTokenFromStorage();
    _privateDirectChatChannel = client?.privateChannel(
      '$directChatChannelName.$chatId',
      authorizationDelegate: EndpointAuthorizableChannelTokenAuthorizationDelegate.forPrivateChannel(
        authorizationEndpoint: Uri.parse('${sl<AppConfig>().baseUrl}/broadcasting/auth'),
        headers: {
          'Authorization': 'Bearer $token',
        },
      ),
    );
  }

  Future<void> initPresenceDirectChatChannel(String chatId) async {
    final token = await TokenStorage.getTokenFromStorage();
    _presenceDirectChatChannel = client?.presenceChannel(
      '$presenceDirectChatChannelName.$chatId',
      authorizationDelegate: EndpointAuthorizableChannelTokenAuthorizationDelegate.forPresenceChannel(
        authorizationEndpoint: Uri.parse('${sl<AppConfig>().baseUrl}/broadcasting/auth'),
        headers: {
          'Authorization': 'Bearer $token',
        },
      ),
    );
  }

  void addPresenceDirectChatListener({
    required int counterpartUserId,
    required void Function(bool isOnline) onCounterpartOnlineChanged,
  }) {
    final channel = _presenceDirectChatChannel;
    if (channel == null) return;

    _presenceDirectChatListeners.add(
      _PresenceDirectChatListener(
        counterpartUserId: counterpartUserId,
        onCounterpartOnlineChanged: onCounterpartOnlineChanged,
      ),
    );

    _bindPresenceChannelEventsIfNeeded();
    channel.subscribe();
    _emitCounterpartPresence(counterpartUserId);
  }

  void removePresenceDirectChatListener(
    void Function(bool isOnline) onCounterpartOnlineChanged,
  ) {
    _presenceDirectChatListeners.removeWhere(
      (listener) => listener.onCounterpartOnlineChanged == onCounterpartOnlineChanged,
    );
    if (_presenceDirectChatListeners.isEmpty) {
      unsubscribePresenceDirectChatChannel();
    }
  }

  void _bindPresenceChannelEventsIfNeeded() {
    if (_presenceChannelEventsBound) return;
    final channel = _presenceDirectChatChannel;
    if (channel == null) return;

    _presenceChannelEventsBound = true;

    _presenceSubscriptionSucceededSubscription =
        channel.whenSubscriptionSucceeded().listen((_) {
      final counterpartIds = _presenceDirectChatListeners
          .map((listener) => listener.counterpartUserId)
          .toSet();
      for (final counterpartUserId in counterpartIds) {
        _emitCounterpartPresence(counterpartUserId);
      }
    });
    _presenceMemberAddedSubscription = channel.whenMemberAdded().listen((event) {
      final memberUserId = _parsePresenceMemberUserId(event.data);
      if (memberUserId == null) return;
      _notifyPresenceListeners(memberUserId, isOnline: true);
    });
    _presenceMemberRemovedSubscription = channel.whenMemberRemoved().listen((event) {
      final memberUserId = _parsePresenceMemberUserId(event.data);
      if (memberUserId == null) return;
      _notifyPresenceListeners(memberUserId, isOnline: false);
    });
  }

  void _emitCounterpartPresence(int counterpartUserId) {
    final channel = _presenceDirectChatChannel;
    if (channel == null) return;
    final members = channel.state?.members;
    if (members == null) return;
    final isOnline = members.getMemberInfo(counterpartUserId.toString()) != null;
    _notifyPresenceListeners(counterpartUserId, isOnline: isOnline);
  }

  void _notifyPresenceListeners(int counterpartUserId, {required bool isOnline}) {
    for (final listener in _presenceDirectChatListeners) {
      if (listener.counterpartUserId == counterpartUserId) {
        listener.onCounterpartOnlineChanged(isOnline);
      }
    }
  }

  int? _parsePresenceMemberUserId(String? data) {
    if (data == null || data.isEmpty) return null;
    try {
      final decoded = jsonDecode(data);
      if (decoded is! Map) return null;
      final rawId = decoded['user_id'] ?? decoded['id'];
      if (rawId == null) return null;
      return int.tryParse(rawId.toString());
    } catch (_) {
      return null;
    }
  }

  void _subscribe() {
    _privateMessagesChannel?.subscribe();
    _privateNotificationChannel?.subscribe();
  }

  void _unsubscribe() {
    _privateMessagesChannel?.unsubscribe();
    _privateNotificationChannel?.unsubscribe();
  }

  void subscribeDirectChatChannel() {
    _privateDirectChatChannel?.subscribe();
  }

  void unsubscribeDirectChatChannel() {
    _privateDirectChatChannel?.unsubscribe();
  }

  void unsubscribePresenceDirectChatChannel() {
    _presenceSubscriptionSucceededSubscription?.cancel();
    _presenceMemberAddedSubscription?.cancel();
    _presenceMemberRemovedSubscription?.cancel();
    _presenceSubscriptionSucceededSubscription = null;
    _presenceMemberAddedSubscription = null;
    _presenceMemberRemovedSubscription = null;
    _presenceDirectChatListeners.clear();
    _presenceChannelEventsBound = false;
    _presenceDirectChatChannel?.unsubscribe();
    _presenceDirectChatChannel = null;
  }

  Future<void> clear() async {
    await client?.disconnect();
    await _lifecycleSubscription?.cancel();
    await _eventSubscription?.cancel();
    await _connectionEstablishedSubscription?.cancel();
    _lifecycleSubscription = null;
    _eventSubscription = null;
    _connectionEstablishedSubscription = null;
    client = null;
    userId = null;
    _privateDirectChatChannel?.unsubscribe();
    unsubscribePresenceDirectChatChannel();
    _privateMessagesChannel?.unsubscribe();
    _privateNotificationChannel?.unsubscribe();
    _privateDirectChatChannel = null;
    _privateMessagesChannel = null;
    _privateNotificationChannel = null;
  }
}

class _PresenceDirectChatListener {
  const _PresenceDirectChatListener({
    required this.counterpartUserId,
    required this.onCounterpartOnlineChanged,
  });

  final int counterpartUserId;
  final void Function(bool isOnline) onCounterpartOnlineChanged;
}