Exchange - WebSockets instead of polling (for User and Chat updates)

lib/core/core_locator.dart

@@ -72,6 +72,7 @@ class CoreLocator {
   }
 
   static void registerServices(GetIt locator) {
+    ExchangeLocator.registerServices(locator);
     MempoolLocator.registerServices(locator);
     SeedLocator.registerServices(locator);
     SwapsLocator.registerServices(locator);

lib/core/exchange/data/datasources/exchange_notification_datasource.dart

@@ -0,0 +1,174 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:bb_mobile/core/exchange/data/datasources/bullbitcoin_api_key_datasource.dart';
+import 'package:bb_mobile/core/utils/logger.dart' show log;
+import 'package:web_socket_channel/io.dart';
+import 'package:web_socket_channel/web_socket_channel.dart';
+
+class ExchangeNotificationDatasource {
+  final String _baseUrl;
+  final BullbitcoinApiKeyDatasource _apiKeyDatasource;
+  final bool _isTestnet;
+
+  WebSocketChannel? _channel;
+  bool _isConnected = false;
+  bool _isManuallyDisconnected = false;
+  bool _isConnecting = false;
+
+  final _messageController = StreamController<Map<String, dynamic>>.broadcast();
+  Stream<Map<String, dynamic>> get messageStream => _messageController.stream;
+
+  bool get isConnected => _isConnected;
+
+  ExchangeNotificationDatasource({
+    required String baseUrl,
+    required BullbitcoinApiKeyDatasource apiKeyDatasource,
+    required bool isTestnet,
+  }) : _baseUrl = baseUrl,
+       _apiKeyDatasource = apiKeyDatasource,
+       _isTestnet = isTestnet;
+
+  String _buildWebSocketUrl() {
+    // Clean the base URL - remove trailing slashes and whitespace
+    var baseUrl = _baseUrl.trim();
+    while (baseUrl.endsWith('/')) {
+      baseUrl = baseUrl.substring(0, baseUrl.length - 1);
+    }
+
+    // Convert HTTP(S) to WS(S) - case insensitive
+    String wsUrl;
+    if (baseUrl.toLowerCase().startsWith('https://')) {
+      wsUrl = 'wss://${baseUrl.substring(8)}';
+    } else if (baseUrl.toLowerCase().startsWith('http://')) {
+      wsUrl = 'ws://${baseUrl.substring(7)}';
+    } else {
+      // Assume wss if no protocol specified
+      wsUrl = baseUrl.startsWith('wss://') || baseUrl.startsWith('ws://')
+          ? baseUrl
+          : 'wss://$baseUrl';
+    }
+
+    // Use /ak/ prefix for API key authenticated endpoint
+    return '$wsUrl/ak/api-commcenter';
+  }
+
+  Future<void> connect() async {
+    if (_isConnected || _isConnecting) {
+      log.info('WebSocket already connected or connecting');
+      return;
+    }
+    _isConnecting = true;
+    _isManuallyDisconnected = false;
+
+    try {
+      // Verify API key exists
+      final apiKey = await _apiKeyDatasource.get(isTestnet: _isTestnet);
+      if (apiKey == null || !apiKey.isActive) {
+        _isConnecting = false;
+        throw Exception('API key not available for WebSocket connection');
+      }
+
+      // Build WebSocket URL
+      final fullUrl = _buildWebSocketUrl();
+      log.info('Base URL: $_baseUrl');
+      log.info('Built WebSocket URL: $fullUrl');
+
+      // Parse and verify URI
+      final uri = Uri.parse(fullUrl);
+      log.info(
+        'Parsed URI - scheme: ${uri.scheme}, host: ${uri.host}, path: ${uri.path}',
+      );
+
+      // Use IOWebSocketChannel to pass X-API-Key header
+      _channel = IOWebSocketChannel.connect(
+        uri,
+        headers: {'X-API-Key': apiKey.key},
+      );
+
+      // Wait for the connection to be ready
+      await _channel!.ready;
+
+      _channel!.stream.listen(
+        _handleMessage,
+        onError: _handleError,
+        onDone: _handleDisconnect,
+      );
+
+      _isConnected = true;
+      _isConnecting = false;
+      log.info('WebSocket connected successfully');
+    } catch (e) {
+      _isConnecting = false;
+      final errorStr = e.toString();
+
+      // Don't auto-reconnect if server doesn't support WebSocket upgrade
+      if (errorStr.contains('not upgraded to websocket')) {
+        log.warning(
+          'WebSocket endpoint not available - server may not support WebSocket on this endpoint. '
+          'Disabling auto-reconnect.',
+        );
+        _isManuallyDisconnected = true; // Prevent auto-reconnect
+      }
+
+      log.severe('WebSocket connection failed: $e');
+      _handleDisconnect();
+      rethrow;
+    }
+  }
+
+  void _handleMessage(dynamic message) {
+    try {
+      final parsed = message is String
+          ? jsonDecode(message) as Map<String, dynamic>
+          : message as Map<String, dynamic>;
+      log.fine('WebSocket message received: $parsed');
+      _messageController.add(parsed);
+    } catch (e) {
+      log.warning('Error parsing WebSocket message: $e');
+    }
+  }
+
+  void _handleDisconnect() {
+    _isConnected = false;
+    _isConnecting = false;
+    log.info('WebSocket disconnected');
+
+    // AUTO-RECONNECT after 5 seconds (unless manually disconnected)
+    if (!_isManuallyDisconnected) {
+      log.info('Scheduling WebSocket auto-reconnect in 5 seconds...');
+      Future.delayed(const Duration(seconds: 5), () {
+        if (!_isManuallyDisconnected && !_isConnecting) {
+          log.info('Attempting WebSocket auto-reconnect...');
+          connect();
+        }
+      });
+    }
+  }
+
+  void _handleError(dynamic error) {
+    log.severe('WebSocket error: $error');
+    _handleDisconnect();
+  }
+
+  void disconnect() {
+    _isManuallyDisconnected = true;
+    _isConnecting = false;
+    _channel?.sink.close();
+    _channel = null;
+    _isConnected = false;
+    log.info('WebSocket manually disconnected');
+  }
+
+  Future<void> reconnect() async {
+    disconnect();
+    _isManuallyDisconnected = false;
+    await Future.delayed(const Duration(seconds: 1));
+    await connect();
+  }
+
+  void dispose() {
+    disconnect();
+    _messageController.close();
+  }
+}

lib/core/exchange/data/models/notification_message_model.dart

@@ -0,0 +1,19 @@
+import 'package:bb_mobile/core/exchange/domain/entity/notification_message.dart';
+
+class NotificationMessageModel {
+  final String type;
+  final String? orderId;
+  final Map<String, dynamic> rawData;
+
+  NotificationMessageModel.fromJson(Map<String, dynamic> json)
+      : type = json['type'] as String? ?? '',
+        orderId = json['orderId'] as String?,
+        rawData = json;
+
+  NotificationMessage toEntity() => NotificationMessage(
+        type: type,
+        orderId: orderId,
+        rawData: rawData,
+      );
+}
+

lib/core/exchange/data/services/exchange_notification_service.dart

@@ -0,0 +1,93 @@
+import 'dart:async';
+
+import 'package:bb_mobile/core/exchange/data/datasources/exchange_notification_datasource.dart';
+import 'package:bb_mobile/core/exchange/data/models/notification_message_model.dart';
+import 'package:bb_mobile/core/exchange/domain/entity/notification_message.dart';
+import 'package:bb_mobile/core/settings/data/settings_repository.dart';
+
+/// Primary/driving adapter that receives WebSocket notification events
+/// and exposes them as a stream for the application to consume.
+///
+/// This service listens to WebSocket events from the exchange API and routes
+/// them based on the current environment (mainnet/testnet).
+class ExchangeNotificationService {
+  final ExchangeNotificationDatasource _mainnetDatasource;
+  final ExchangeNotificationDatasource _testnetDatasource;
+  final SettingsRepository _settingsRepository;
+
+  final StreamController<NotificationMessage> _messageController =
+      StreamController<NotificationMessage>.broadcast();
+
+  StreamSubscription<Map<String, dynamic>>? _mainnetSubscription;
+  StreamSubscription<Map<String, dynamic>>? _testnetSubscription;
+  bool _isTestnet = false;
+
+  ExchangeNotificationService({
+    required ExchangeNotificationDatasource mainnetDatasource,
+    required ExchangeNotificationDatasource testnetDatasource,
+    required SettingsRepository settingsRepository,
+  }) : _mainnetDatasource = mainnetDatasource,
+       _testnetDatasource = testnetDatasource,
+       _settingsRepository = settingsRepository {
+    _setupInternalListeners();
+  }
+
+  void _setupInternalListeners() {
+    // Listen to mainnet and forward messages only when on mainnet
+    _mainnetSubscription = _mainnetDatasource.messageStream.listen((json) {
+      if (!_isTestnet) {
+        final entity = NotificationMessageModel.fromJson(json).toEntity();
+        _messageController.add(entity);
+      }
+    });
+
+    // Listen to testnet and forward messages only when on testnet
+    _testnetSubscription = _testnetDatasource.messageStream.listen((json) {
+      if (_isTestnet) {
+        final entity = NotificationMessageModel.fromJson(json).toEntity();
+        _messageController.add(entity);
+      }
+    });
+  }
+
+  Future<ExchangeNotificationDatasource> _getDatasource() async {
+    final settings = await _settingsRepository.fetch();
+    _isTestnet = settings.environment.isTestnet;
+    return _isTestnet ? _testnetDatasource : _mainnetDatasource;
+  }
+
+  /// Connect to the WebSocket for the current environment (mainnet/testnet)
+  Future<void> connect() async {
+    final datasource = await _getDatasource();
+    await datasource.connect();
+  }
+
+  /// Disconnect from both mainnet and testnet WebSockets
+  void disconnect() {
+    _mainnetDatasource.disconnect();
+    _testnetDatasource.disconnect();
+  }
+
+  /// Single stream of notification messages from the active network
+  Stream<NotificationMessage> get messageStream => _messageController.stream;
+
+  /// Check if the current environment's WebSocket is connected
+  Future<bool> get isConnected async {
+    final datasource = await _getDatasource();
+    return datasource.isConnected;
+  }
+
+  /// Reconnect to the WebSocket for the current environment
+  /// Call this when the network changes to switch connections
+  Future<void> reconnect() async {
+    disconnect();
+    await Future.delayed(const Duration(milliseconds: 500));
+    await connect();
+  }
+
+  void dispose() {
+    _mainnetSubscription?.cancel();
+    _testnetSubscription?.cancel();
+    _messageController.close();
+  }
+}

lib/core/exchange/domain/entity/notification_message.dart

@@ -0,0 +1,12 @@
+class NotificationMessage {
+  final String type; // 'user', 'message', 'order'
+  final String? orderId; // For order-specific messages
+  final Map<String, dynamic> rawData;
+
+  const NotificationMessage({
+    required this.type,
+    this.orderId,
+    required this.rawData,
+  });
+}
+

lib/core/exchange/exchange_locator.dart

@@ -1,5 +1,6 @@
 import 'package:bb_mobile/core/exchange/data/datasources/bullbitcoin_api_datasource.dart';
 import 'package:bb_mobile/core/exchange/data/datasources/bullbitcoin_api_key_datasource.dart';
+import 'package:bb_mobile/core/exchange/data/datasources/exchange_notification_datasource.dart';
 import 'package:bb_mobile/core/exchange/data/datasources/exchange_support_chat_datasource.dart';
 import 'package:bb_mobile/core/exchange/data/datasources/price_local_datasource.dart';
 import 'package:bb_mobile/core/exchange/data/datasources/price_remote_datasource.dart';
@@ -10,6 +11,7 @@ import 'package:bb_mobile/core/exchange/data/repository/exchange_rate_repository
 import 'package:bb_mobile/core/exchange/data/repository/exchange_support_chat_repository_impl.dart';
 import 'package:bb_mobile/core/exchange/data/repository/exchange_user_repository_impl.dart';
 import 'package:bb_mobile/core/exchange/data/repository/price_repository_impl.dart';
+import 'package:bb_mobile/core/exchange/data/services/exchange_notification_service.dart';
 import 'package:bb_mobile/core/exchange/domain/repositories/exchange_api_key_repository.dart';
 import 'package:bb_mobile/core/exchange/domain/repositories/exchange_funding_repository.dart';
 import 'package:bb_mobile/core/exchange/domain/repositories/exchange_order_repository.dart';
@@ -21,9 +23,9 @@ import 'package:bb_mobile/core/exchange/domain/usecases/convert_currency_to_sats
 import 'package:bb_mobile/core/exchange/domain/usecases/convert_sats_to_currency_amount_usecase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/create_log_attachment_usecase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/delete_exchange_api_key_usecase.dart';
+import 'package:bb_mobile/core/exchange/domain/usecases/get_announcements_usecase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/get_available_currencies_usecase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/get_exchange_funding_details_usecase.dart';
-import 'package:bb_mobile/core/exchange/domain/usecases/get_announcements_usecase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/get_exchange_user_summary_usecase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/get_order_usercase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/get_price_history_usecase.dart';
@@ -33,8 +35,8 @@ import 'package:bb_mobile/core/exchange/domain/usecases/label_exchange_orders_us
 import 'package:bb_mobile/core/exchange/domain/usecases/list_all_orders_usecase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/refresh_price_history_usecase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/save_exchange_api_key_usecase.dart';
-import 'package:bb_mobile/core/exchange/domain/usecases/send_support_chat_message_usecase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/save_user_preferences_usecase.dart';
+import 'package:bb_mobile/core/exchange/domain/usecases/send_support_chat_message_usecase.dart';
 import 'package:bb_mobile/core/labels/data/label_datasource.dart';
 import 'package:bb_mobile/core/labels/data/label_repository.dart';
 import 'package:bb_mobile/core/labels/domain/batch_labels_usecase.dart';
@@ -109,6 +111,25 @@ class ExchangeLocator {
       ),
       instanceName: 'testnetExchangeSupportChatDatasource',
     );
+
+    // WebSocket Notification Datasources
+    locator.registerLazySingleton<ExchangeNotificationDatasource>(
+      () => ExchangeNotificationDatasource(
+        baseUrl: ApiServiceConstants.bbApiUrl,
+        apiKeyDatasource: locator<BullbitcoinApiKeyDatasource>(),
+        isTestnet: false,
+      ),
+      instanceName: 'mainnetExchangeNotificationDatasource',
+    );
+
+    locator.registerLazySingleton<ExchangeNotificationDatasource>(
+      () => ExchangeNotificationDatasource(
+        baseUrl: ApiServiceConstants.bbApiTestUrl,
+        apiKeyDatasource: locator<BullbitcoinApiKeyDatasource>(),
+        isTestnet: true,
+      ),
+      instanceName: 'testnetExchangeNotificationDatasource',
+    );
   }
 
   static void registerRepositories(GetIt locator) {
@@ -489,9 +510,25 @@ class ExchangeLocator {
     );
   }
 
+  static void registerServices(GetIt locator) {
+    // WebSocket Notification Service (primary/driving adapter)
+    locator.registerLazySingleton<ExchangeNotificationService>(
+      () => ExchangeNotificationService(
+        mainnetDatasource: locator<ExchangeNotificationDatasource>(
+          instanceName: 'mainnetExchangeNotificationDatasource',
+        ),
+        testnetDatasource: locator<ExchangeNotificationDatasource>(
+          instanceName: 'testnetExchangeNotificationDatasource',
+        ),
+        settingsRepository: locator<SettingsRepository>(),
+      ),
+    );
+  }
+
   static void setup(GetIt locator) {
     registerDatasources(locator);
     registerRepositories(locator);
     registerUseCases(locator);
+    registerServices(locator);
   }
 }