Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
315 changes: 140 additions & 175 deletions mobile/lib/controllers/auth_state.dart

Large diffs are not rendered by default.

898 changes: 355 additions & 543 deletions mobile/lib/controllers/chat/active_chat_controller.dart

Large diffs are not rendered by default.

159 changes: 84 additions & 75 deletions mobile/lib/controllers/chat/chat_connection_controller.dart
Original file line number Diff line number Diff line change
@@ -1,12 +1,15 @@
import 'dart:async';
import 'dart:convert';
import 'dart:math';
import 'package:flutter/material.dart';
import 'package:connectivity_plus/connectivity_plus.dart';
import 'package:mobile/services/chat/chat_event_handler.dart';
import 'package:mobile/services/chat/chat_sync_service.dart';
import '../../services/ws_service.dart';
import '../../services/auth_service.dart';

enum ConnectionStatus { connected, connecting, offline }

class ChatConnectionController extends ChangeNotifier
with WidgetsBindingObserver {
final WebSocketService _ws = WebSocketService();
Expand All @@ -15,7 +18,11 @@ class ChatConnectionController extends ChangeNotifier
final ChatEventHandler eventHandler;

bool isOffline = false;
ConnectionStatus status = ConnectionStatus.connecting;

bool _isWsConnecting = false;
bool _isPaused = false;
int _failedAttempts = 0;
Timer? _reconnectTimer;
StreamSubscription? _connectivitySubscription;
StreamSubscription? _wsSubscription;
Expand All @@ -27,62 +34,63 @@ class ChatConnectionController extends ChangeNotifier
result,
) {
final bool currentlyOffline = result.contains(ConnectivityResult.none);

if (isOffline != currentlyOffline) {
isOffline = currentlyOffline;
notifyListeners();

if (!isOffline) {
connectWebSocket();
eventHandler.inboxController.loadInbox();
} else {
eventHandler.activeChatController.isPeerOnline = false;
eventHandler.activeChatController.isPeerTyping = false;
eventHandler.activeChatController.refreshUI();

_ws.disconnect();
}
if (isOffline == currentlyOffline) return;

isOffline = currentlyOffline;
if (isOffline) {
_markPeerOffline();
_wsSubscription?.cancel();
_ws.disconnect();
_setStatus(ConnectionStatus.offline);
} else {
_failedAttempts = 0;
connectWebSocket();
eventHandler.inboxController.loadInbox();
}
});
}

void _setStatus(ConnectionStatus next) {
if (status != next) {
status = next;
notifyListeners();
}
}

void _markPeerOffline() {
final chat = eventHandler.activeChatController;
if (chat.isPeerOnline || chat.isPeerTyping) {
chat.isPeerOnline = false;
chat.isPeerTyping = false;
chat.refreshUI();
}
}

Future<void> connectWebSocket() async {
if (_ws.isConnected || _isWsConnecting) return;
if (_ws.isConnected || _isWsConnecting || isOffline || _isPaused) return;
_isWsConnecting = true;
_reconnectTimer?.cancel();
_setStatus(ConnectionStatus.connecting);

try {
_reconnectTimer?.cancel();

final token = await _auth.getToken();
if (token == null) {
_isWsConnecting = false;
return;
}
// Access tokens live 15 minutes, so refresh before every handshake.
final token = await _auth.getValidAccessToken();
if (token == null) return;

await _wsSubscription?.cancel();
_ws.disconnect();

final connected = await _ws.connect(token);
if (!connected) {
_triggerReconnectLoop();
if (!await _ws.connect(token)) {
_scheduleReconnect();
return;
}

_syncService.processOfflineQueue(
eventHandler.currentUserId,
activeChatController: eventHandler.activeChatController,
);

final currentChatId = eventHandler.activeChatController.currentChatUserId;
if (currentChatId != null &&
!eventHandler.activeChatController.isCurrentChatGroup) {
_ws.sendRequestStatus(targetId: currentChatId);
}
_failedAttempts = 0;
_setStatus(ConnectionStatus.connected);

_wsSubscription = _ws.stream?.listen(
(rawFrame) {
try {
debugPrint("📥 WS Received: $rawFrame");
final decoded = jsonDecode(rawFrame);
if (decoded is Map<String, dynamic>) {
eventHandler.handleIncomingEvent(decoded);
Expand All @@ -92,78 +100,79 @@ class ChatConnectionController extends ChangeNotifier
}
},
onError: (err) {
debugPrint("WS Pipeline Error: $err");
_triggerReconnectLoop();
debugPrint("WS error: $err");
_scheduleReconnect();
},
onDone: () {
debugPrint("WS Pipeline Closed by Server.");
_triggerReconnectLoop();
debugPrint("WS closed");
_scheduleReconnect();
},
cancelOnError: true,
);

_syncService.processOfflineQueue(
eventHandler.currentUserId,
activeChatController: eventHandler.activeChatController,
);

final chat = eventHandler.activeChatController;
if (chat.currentChatUserId != null && !chat.isCurrentChatGroup) {
_ws.sendRequestStatus(targetId: chat.currentChatUserId!);
}
} catch (e) {
debugPrint("WS Setup Error: $e");
_triggerReconnectLoop();
debugPrint("WS setup error: $e");
_scheduleReconnect();
} finally {
_isWsConnecting = false;
}
_ws.onConnectionLost = () {
debugPrint(
"ConnectionController: Caught immediate socket loss from watchdog.",
);
_triggerReconnectLoop();
};
}

void _triggerReconnectLoop() {
void _scheduleReconnect() {
_ws.disconnect();
_isWsConnecting = false;
_markPeerOffline();
if (isOffline || _isPaused) return;

if (eventHandler.activeChatController.isPeerOnline) {
eventHandler.activeChatController.isPeerOnline = false;
eventHandler.activeChatController.refreshUI();
}
_setStatus(ConnectionStatus.connecting);
_failedAttempts++;
final seconds = min(30, 2 * pow(2, min(_failedAttempts - 1, 4)).toInt());

_reconnectTimer?.cancel();
_reconnectTimer = Timer(const Duration(seconds: 4), () {
if (!isOffline) {
connectWebSocket();
}
});
_reconnectTimer = Timer(Duration(seconds: seconds), connectWebSocket);
}

/// Reconnect immediately, e.g. from a "Retry" button.
void retryNow() {
_failedAttempts = 0;
connectWebSocket();
}

void disconnectWebSocket() {
_reconnectTimer?.cancel();
_wsSubscription?.cancel();
_ws.disconnect();
}

@override
void didChangeAppLifecycleState(AppLifecycleState state) {
if (state == AppLifecycleState.resumed) {
_syncService.processOfflineQueue(
eventHandler.currentUserId,
activeChatController: eventHandler.activeChatController,
);

_isPaused = false;
connectWebSocket();

Future.delayed(const Duration(milliseconds: 500), () {
if (!isOffline) {
eventHandler.inboxController.loadInbox();
}
});
if (!isOffline) {
eventHandler.inboxController.loadInbox();
eventHandler.activeChatController.syncActiveChatSilently();
}
} else if (state == AppLifecycleState.paused) {
_reconnectTimer?.cancel();
_ws.disconnect();
_isPaused = true;
disconnectWebSocket();
}
}

@override
void dispose() {
WidgetsBinding.instance.removeObserver(this);
_connectivitySubscription?.cancel();
_wsSubscription?.cancel();
_reconnectTimer?.cancel();
_ws.disconnect();
disconnectWebSocket();
super.dispose();
}
}
90 changes: 66 additions & 24 deletions mobile/lib/controllers/chat/chat_search_controller.dart
Original file line number Diff line number Diff line change
@@ -1,16 +1,35 @@
import 'package:flutter/material.dart';
import '../../core/constants.dart';
import '../../models/user.dart';
import '../../services/api_services.dart';

/// User directory search (`GET /api/users/search`, rate limited server-side).
class ChatSearchController extends ChangeNotifier {
final ApiService _api = ApiService();

List<dynamic> contactSearchResults = [];
List<UserModel> results = [];
bool isSearchLoading = false;
bool isLoadingMore = false;
bool hasMore = false;
String? error;

String _query = '';
int _page = 1;
int _requestId = 0;

static const int minQueryLength = 2;

Future<void> queryUsers(String term) async {
final String cleanTerm = term.trim();
if (cleanTerm.isEmpty || cleanTerm.length < 3) {
contactSearchResults.clear();
final query = term.trim();
final requestId = ++_requestId;
_query = query;
_page = 1;
error = null;

if (query.length < minQueryLength) {
results = [];
hasMore = false;
isSearchLoading = false;
notifyListeners();
return;
}
Expand All @@ -19,34 +38,57 @@ class ChatSearchController extends ChangeNotifier {
notifyListeners();

try {
final res = await _api.searchUsers(term);
contactSearchResults = _extractDataList(res.data, ['users']);
final page = await _fetch(query, 1);
if (requestId != _requestId) return;
results = page;
hasMore = page.length >= ServerLimits.searchPageSize;
} catch (e) {
debugPrint("User query failure: $e");
contactSearchResults.clear();
if (requestId != _requestId) return;
results = [];
error = ApiService.errorMessage(e, fallback: "Search failed.");
} finally {
isSearchLoading = false;
notifyListeners();
if (requestId == _requestId) {
isSearchLoading = false;
notifyListeners();
}
}
}

void clearSearch() {
if (contactSearchResults.isNotEmpty || isSearchLoading) {
contactSearchResults.clear();
isSearchLoading = false;
Future<void> loadMore() async {
if (!hasMore || isLoadingMore || isSearchLoading) return;
final requestId = _requestId;
isLoadingMore = true;
notifyListeners();

try {
final page = await _fetch(_query, _page + 1);
if (requestId != _requestId) return;
_page++;
final known = results.map((u) => u.id).toSet();
results = [...results, ...page.where((u) => !known.contains(u.id))];
hasMore = page.length >= ServerLimits.searchPageSize;
} catch (e) {
error = ApiService.errorMessage(e, fallback: "Search failed.");
} finally {
isLoadingMore = false;
notifyListeners();
}
}

List<dynamic> _extractDataList(dynamic data, List<String> fallbackKeys) {
if (data == null) return [];
if (data is List) return data;
if (data is Map) {
if (data['data'] is List) return data['data'];
for (final key in fallbackKeys) {
if (data[key] is List) return data[key];
}
}
return [];
Future<List<UserModel>> _fetch(String query, int page) async {
final res = await _api.searchUsers(query, page: page);
return ApiService.dataList(res.data)
.map((json) => UserModel.fromJson(Map<String, dynamic>.from(json)))
.toList();
}

void clearSearch() {
_requestId++;
_query = '';
results = [];
hasMore = false;
error = null;
isSearchLoading = false;
notifyListeners();
}
}
Loading
Loading