Files
iiEsaywebUIapp/lib/core/services/socket_service.dart

519 lines
14 KiB
Dart
Raw Normal View History

2025-09-26 01:38:00 +05:30
import 'package:flutter/widgets.dart';
import 'package:socket_io_client/socket_io_client.dart' as io;
2025-09-26 01:38:00 +05:30
2025-08-31 14:02:44 +05:30
import '../models/server_config.dart';
import 'socket_tls_override.dart';
2025-09-25 22:36:42 +05:30
2025-09-26 01:38:00 +05:30
typedef SocketChatEventHandler =
void Function(
Map<String, dynamic> event,
void Function(dynamic response)? ack,
);
typedef SocketChannelEventHandler =
void Function(
Map<String, dynamic> event,
void Function(dynamic response)? ack,
);
class SocketService with WidgetsBindingObserver {
2025-08-31 14:02:44 +05:30
final ServerConfig serverConfig;
final bool websocketOnly;
final bool allowWebsocketUpgrade;
io.Socket? _socket;
2025-09-23 13:43:01 +05:30
String? _authToken;
2025-09-26 01:38:00 +05:30
bool _isAppForeground = true;
final Map<String, _ChatEventRegistration> _chatEventHandlers = {};
final Map<String, _ChannelEventRegistration> _channelEventHandlers = {};
int _handlerSeed = 0;
2025-08-31 14:02:44 +05:30
SocketService({
required this.serverConfig,
2025-09-23 13:43:01 +05:30
String? authToken,
this.websocketOnly = false,
this.allowWebsocketUpgrade = true,
2025-09-26 01:38:00 +05:30
}) : _authToken = authToken {
final binding = WidgetsBinding.instance;
final lifecycle = binding.lifecycleState;
_isAppForeground =
lifecycle == null || lifecycle == AppLifecycleState.resumed;
binding.addObserver(this);
}
@override
void didChangeAppLifecycleState(AppLifecycleState state) {
_isAppForeground = state == AppLifecycleState.resumed;
}
2025-08-31 14:02:44 +05:30
String? get sessionId => _socket?.id;
io.Socket? get socket => _socket;
2025-09-23 13:43:01 +05:30
String? get authToken => _authToken;
2025-08-31 14:02:44 +05:30
bool get isConnected => _socket?.connected == true;
2025-09-26 01:38:00 +05:30
bool get isAppForeground => _isAppForeground;
2025-08-31 14:02:44 +05:30
Future<void> connect({bool force = false}) async {
if (_socket != null && _socket!.connected && !force) return;
try {
_socket?.dispose();
} catch (_) {}
2025-09-07 14:40:20 +05:30
String base = serverConfig.url.replaceFirst(RegExp(r'/+$'), '');
// Normalize accidental ":0" ports or invalid port values in stored URL
try {
final u = Uri.parse(base);
if (u.hasPort && u.port == 0) {
// Drop the explicit :0 to fall back to scheme default (80/443)
base = '${u.scheme}://${u.host}${u.path.isEmpty ? '' : u.path}';
}
} catch (_) {}
2025-08-31 14:02:44 +05:30
final path = '/ws/socket.io';
final usePollingOnly = !websocketOnly && !allowWebsocketUpgrade;
final transports = websocketOnly
? const ['websocket']
: usePollingOnly
? const ['polling']
: const ['polling', 'websocket'];
final builder = io.OptionBuilder()
// Transport selection switches between WebSocket-only and polling fallback
.setTransports(transports)
.setRememberUpgrade(!websocketOnly && allowWebsocketUpgrade)
.setUpgrade(!websocketOnly && allowWebsocketUpgrade)
// Tune reconnect/backoff and timeouts
.setReconnectionAttempts(0) // 0/Infinity semantics: unlimited attempts
.setReconnectionDelay(1000)
.setReconnectionDelayMax(5000)
.setRandomizationFactor(0.5)
.setTimeout(20000)
.setPath(path);
2025-09-08 13:18:38 +05:30
// Merge Authorization (if any) with user-defined custom headers for the
// Socket.IO handshake. Avoid overriding reserved headers.
final Map<String, String> extraHeaders = {};
2025-09-23 13:43:01 +05:30
if (_authToken != null && _authToken!.isNotEmpty) {
extraHeaders['Authorization'] = 'Bearer $_authToken';
builder.setAuth({'token': _authToken});
2025-09-08 13:18:38 +05:30
}
if (serverConfig.customHeaders.isNotEmpty) {
final reserved = {
'authorization',
'content-type',
'accept',
// Socket/WebSocket reserved or managed by client/runtime
'host',
'origin',
'connection',
'upgrade',
'sec-websocket-key',
'sec-websocket-version',
'sec-websocket-extensions',
'sec-websocket-protocol',
};
serverConfig.customHeaders.forEach((key, value) {
final lower = key.toLowerCase();
if (!reserved.contains(lower) && value.isNotEmpty) {
// Do not overwrite Authorization we already set from authToken
2025-09-23 13:43:01 +05:30
if (lower == 'authorization' &&
extraHeaders.containsKey('Authorization')) {
2025-09-08 13:18:38 +05:30
return;
}
extraHeaders[key] = value;
}
});
}
if (extraHeaders.isNotEmpty) {
builder.setExtraHeaders(extraHeaders);
}
_socket = createSocketWithOptionalBadCertOverride(
base,
builder,
serverConfig,
);
2025-08-31 14:02:44 +05:30
2025-09-26 01:38:00 +05:30
_bindCoreSocketHandlers();
2025-08-31 14:02:44 +05:30
}
2025-09-23 13:43:01 +05:30
/// Update the auth token used by the socket service.
/// If connected, emits a best-effort rejoin with the new token.
void updateAuthToken(String? token) {
_authToken = token;
if (_socket?.connected == true &&
_authToken != null &&
_authToken!.isNotEmpty) {
try {
_socket!.emit('user-join', {
'auth': {'token': _authToken},
});
} catch (_) {}
}
}
2025-09-26 01:38:00 +05:30
SocketEventSubscription addChatEventHandler({
String? conversationId,
String? sessionId,
bool requireFocus = true,
required SocketChatEventHandler handler,
}) {
final id = _nextHandlerId();
_chatEventHandlers[id] = _ChatEventRegistration(
id: id,
conversationId: conversationId,
sessionId: sessionId,
requireFocus: requireFocus,
handler: handler,
);
return SocketEventSubscription(() => _chatEventHandlers.remove(id));
2025-08-31 14:02:44 +05:30
}
2025-09-26 01:38:00 +05:30
SocketEventSubscription addChannelEventHandler({
String? conversationId,
String? sessionId,
bool requireFocus = true,
required SocketChannelEventHandler handler,
}) {
final id = _nextHandlerId();
_channelEventHandlers[id] = _ChannelEventRegistration(
id: id,
conversationId: conversationId,
sessionId: sessionId,
requireFocus: requireFocus,
handler: handler,
);
return SocketEventSubscription(() => _channelEventHandlers.remove(id));
2025-09-01 16:28:49 +05:30
}
2025-09-26 01:38:00 +05:30
void clearChatEventHandlers() {
_chatEventHandlers.clear();
2025-08-31 14:02:44 +05:30
}
2025-09-26 01:38:00 +05:30
void clearChannelEventHandlers() {
_channelEventHandlers.clear();
2025-09-01 16:28:49 +05:30
}
// Subscribe to an arbitrary socket.io event (used for dynamic tool channels)
void onEvent(String eventName, void Function(dynamic data) handler) {
_socket?.on(eventName, handler);
}
void offEvent(String eventName) {
_socket?.off(eventName);
}
2025-09-23 13:43:01 +05:30
2025-08-31 14:02:44 +05:30
void dispose() {
try {
_socket?.dispose();
} catch (_) {}
_socket = null;
2025-09-26 01:38:00 +05:30
WidgetsBinding.instance.removeObserver(this);
_chatEventHandlers.clear();
_channelEventHandlers.clear();
2025-08-31 14:02:44 +05:30
}
2025-09-07 22:37:52 +05:30
// Best-effort: ensure there is an active connection and wait briefly.
// Returns true if connected by the end of the timeout.
2025-09-23 13:43:01 +05:30
Future<bool> ensureConnected({
Duration timeout = const Duration(seconds: 2),
}) async {
2025-09-07 22:37:52 +05:30
if (isConnected) return true;
try {
await connect();
} catch (_) {}
final start = DateTime.now();
while (!isConnected && DateTime.now().difference(start) < timeout) {
await Future.delayed(const Duration(milliseconds: 50));
}
return isConnected;
}
2025-09-26 01:38:00 +05:30
void _bindCoreSocketHandlers() {
final socket = _socket;
if (socket == null) return;
socket
..off('events', _handleChatEvent)
2025-09-26 01:38:00 +05:30
..off('chat-events', _handleChatEvent)
..off('events:channel', _handleChannelEvent)
2025-09-26 01:38:00 +05:30
..off('channel-events', _handleChannelEvent)
..off('connect', _handleConnect)
..off('connect_error', _handleConnectError)
..off('reconnect_attempt', _handleReconnectAttempt)
..off('reconnect', _handleReconnect)
..off('reconnect_failed', _handleReconnectFailed)
..off('disconnect', _handleDisconnect);
socket
..on('events', _handleChatEvent)
2025-09-26 01:38:00 +05:30
..on('chat-events', _handleChatEvent)
..on('events:channel', _handleChannelEvent)
2025-09-26 01:38:00 +05:30
..on('channel-events', _handleChannelEvent)
..on('connect', _handleConnect)
..on('connect_error', _handleConnectError)
..on('reconnect_attempt', _handleReconnectAttempt)
..on('reconnect', _handleReconnect)
..on('reconnect_failed', _handleReconnectFailed)
..on('disconnect', _handleDisconnect);
}
void _handleConnect(dynamic _) {
if (_authToken != null && _authToken!.isNotEmpty) {
_socket?.emit('user-join', {
'auth': {'token': _authToken},
});
}
}
void _handleReconnectAttempt(dynamic attempt) {
// Silent reconnection attempt
2025-09-26 01:38:00 +05:30
}
void _handleReconnect(dynamic attempt) {
if (_authToken != null && _authToken!.isNotEmpty) {
_socket?.emit('user-join', {
'auth': {'token': _authToken},
});
}
}
void _handleConnectError(dynamic err) {}
2025-09-26 01:38:00 +05:30
void _handleReconnectFailed(dynamic _) {}
2025-09-26 01:38:00 +05:30
void _handleDisconnect(dynamic reason) {
// Silent disconnect
2025-09-26 01:38:00 +05:30
}
void _handleChatEvent(dynamic data, [dynamic ack]) {
final map = _coerceToMap(data);
if (map == null) return;
final ackFn = _wrapAck(ack);
final sessionId = _extractSessionId(map);
final chatId = map['chat_id']?.toString();
2025-09-26 20:57:54 +05:30
final channelId = _extractChannelId(map);
2025-09-26 01:38:00 +05:30
for (final registration in List<_ChatEventRegistration>.from(
_chatEventHandlers.values,
)) {
if (!_shouldDeliver(
registration.conversationId,
registration.sessionId,
chatId,
sessionId,
registration.requireFocus,
2025-09-26 20:57:54 +05:30
incomingChannelId: channelId,
2025-09-26 01:38:00 +05:30
)) {
continue;
}
try {
registration.handler(map, ackFn);
} catch (_) {}
}
}
void _handleChannelEvent(dynamic data, [dynamic ack]) {
final map = _coerceToMap(data);
if (map == null) return;
final ackFn = _wrapAck(ack);
final sessionId = _extractSessionId(map);
final chatId = map['chat_id']?.toString();
2025-09-26 20:57:54 +05:30
final channelId = _extractChannelId(map);
2025-09-26 01:38:00 +05:30
for (final registration in List<_ChannelEventRegistration>.from(
_channelEventHandlers.values,
)) {
if (!_shouldDeliver(
registration.conversationId,
registration.sessionId,
chatId,
sessionId,
registration.requireFocus,
2025-09-26 20:57:54 +05:30
incomingChannelId: channelId,
2025-09-26 01:38:00 +05:30
)) {
continue;
}
try {
registration.handler(map, ackFn);
} catch (_) {}
}
}
bool _shouldDeliver(
String? registeredConversationId,
String? registeredSessionId,
String? incomingConversationId,
String? incomingSessionId,
2025-09-26 20:57:54 +05:30
bool requireFocus, {
String? incomingChannelId,
}) {
final matchesConversation =
2025-09-26 01:38:00 +05:30
registeredConversationId == null ||
(incomingConversationId != null &&
2025-09-26 20:57:54 +05:30
registeredConversationId == incomingConversationId) ||
(incomingChannelId != null &&
registeredConversationId == incomingChannelId);
2025-09-26 01:38:00 +05:30
final matchesSession =
registeredSessionId != null &&
incomingSessionId != null &&
registeredSessionId == incomingSessionId;
2025-09-26 20:57:54 +05:30
if (!matchesConversation && !matchesSession) {
2025-09-26 01:38:00 +05:30
return false;
}
if (!requireFocus) {
return true;
}
if (matchesSession) {
// Session-targeted messages should always pass through even if unfocused
return true;
}
return _isAppForeground;
}
Map<String, dynamic>? _coerceToMap(dynamic data) {
if (data is Map<String, dynamic>) {
return data;
}
if (data is Map) {
return Map<String, dynamic>.from(data);
}
return null;
}
void Function(dynamic response)? _wrapAck(dynamic ack) {
if (ack is! Function) return null;
return (dynamic payload) {
try {
if (payload is List) {
Function.apply(ack, payload);
} else if (payload == null) {
Function.apply(ack, const []);
} else {
Function.apply(ack, [payload]);
}
} catch (_) {}
};
}
String? _extractSessionId(Map<String, dynamic> event) {
String? candidate;
if (event['session_id'] != null) {
candidate = event['session_id'].toString();
}
final data = event['data'];
if (data is Map) {
if (candidate == null && data['session_id'] != null) {
candidate = data['session_id'].toString();
}
if (candidate == null && data['sessionId'] != null) {
candidate = data['sessionId'].toString();
}
final inner = data['data'];
if (inner is Map) {
if (candidate == null && inner['session_id'] != null) {
candidate = inner['session_id'].toString();
}
if (candidate == null && inner['sessionId'] != null) {
candidate = inner['sessionId'].toString();
}
}
}
2025-09-26 20:57:54 +05:30
return candidate;
}
String? _extractChannelId(Map<String, dynamic> event) {
String? candidate;
if (event['channel_id'] != null) {
candidate = event['channel_id'].toString();
}
if (candidate == null && event['channelId'] != null) {
candidate = event['channelId'].toString();
}
final data = event['data'];
if (data is Map) {
if (candidate == null && data['channel_id'] != null) {
candidate = data['channel_id'].toString();
}
if (candidate == null && data['channelId'] != null) {
candidate = data['channelId'].toString();
}
final inner = data['data'];
if (inner is Map) {
if (candidate == null && inner['channel_id'] != null) {
candidate = inner['channel_id'].toString();
}
if (candidate == null && inner['channelId'] != null) {
candidate = inner['channelId'].toString();
}
}
}
2025-09-26 01:38:00 +05:30
return candidate;
}
String _nextHandlerId() {
_handlerSeed += 1;
return _handlerSeed.toString();
}
}
class SocketEventSubscription {
SocketEventSubscription(this._dispose);
final VoidCallback _dispose;
bool _isDisposed = false;
void dispose() {
if (_isDisposed) return;
_isDisposed = true;
_dispose();
}
}
class _ChatEventRegistration {
_ChatEventRegistration({
required this.id,
required this.handler,
this.conversationId,
this.sessionId,
this.requireFocus = true,
});
final String id;
final String? conversationId;
final String? sessionId;
final bool requireFocus;
final SocketChatEventHandler handler;
}
class _ChannelEventRegistration {
_ChannelEventRegistration({
required this.id,
required this.handler,
this.conversationId,
this.sessionId,
this.requireFocus = true,
});
final String id;
final String? conversationId;
final String? sessionId;
final bool requireFocus;
final SocketChannelEventHandler handler;
2025-08-31 14:02:44 +05:30
}