545 lines
14 KiB
Dart
545 lines
14 KiB
Dart
// lib/remote/remote_ws_client.dart
|
|
//
|
|
// Patched version with small robustness improvements, extra debug logging,
|
|
// clearer backoff/reconnect behavior, need_recovery support,
|
|
// and readiness/disconnection callbacks for RemoteController.
|
|
|
|
import 'dart:async';
|
|
import 'dart:convert';
|
|
|
|
import 'package:flutter/foundation.dart';
|
|
import 'package:web_socket_channel/io.dart';
|
|
import 'package:web_socket_channel/web_socket_channel.dart';
|
|
|
|
import 'remote_added_queue.dart';
|
|
import 'remote_state_store.dart';
|
|
|
|
typedef WsEventHandler = Future<bool> Function(Map<String, dynamic> msg);
|
|
typedef WsRecoveryHandler = Future<void> Function();
|
|
|
|
class RemoteWsClient {
|
|
RemoteWsClient({
|
|
required this.wsUrl,
|
|
required this.store,
|
|
required this.deduper,
|
|
required this.onEvent,
|
|
required this.onNeedRecoverySync,
|
|
this.onReady,
|
|
this.onDisconnected,
|
|
this.autoReconnect = true,
|
|
this.reconnectDelay = const Duration(milliseconds: 1500),
|
|
this.maxReconnectDelay = const Duration(seconds: 30),
|
|
this.dormantWindow = const Duration(minutes: 2),
|
|
this.pingInterval = const Duration(seconds: 25),
|
|
this.needFullSyncDelay = const Duration(milliseconds: 800),
|
|
});
|
|
|
|
final String wsUrl;
|
|
final RemoteStateStore store;
|
|
final EventRingDeduper deduper;
|
|
|
|
/// Deve ritornare true SOLO quando l'evento è applicato con successo
|
|
/// oppure quando hai schedulato un lavoro durable che completerai sicuramente.
|
|
final WsEventHandler onEvent;
|
|
|
|
/// Trigger async per fare una sync di recovery progressiva.
|
|
final WsRecoveryHandler onNeedRecoverySync;
|
|
|
|
/// Chiamato quando il WS è autenticato e il client può mostrare i remoti.
|
|
///
|
|
/// RemoteController userà questo callback per impostare:
|
|
/// RemoteSyncState.upToDate.
|
|
final VoidCallback? onReady;
|
|
|
|
/// Chiamato quando il WS cade o non riesce l'autenticazione.
|
|
///
|
|
/// RemoteController userà questo callback per impostare:
|
|
/// RemoteSyncState.serverDown e nascondere i remoti.
|
|
final VoidCallback? onDisconnected;
|
|
|
|
final bool autoReconnect;
|
|
final Duration reconnectDelay;
|
|
final Duration maxReconnectDelay;
|
|
final Duration dormantWindow;
|
|
final Duration pingInterval;
|
|
final Duration needFullSyncDelay;
|
|
|
|
WebSocketChannel? _ch;
|
|
StreamSubscription? _sub;
|
|
|
|
bool _connecting = false;
|
|
bool _closing = false;
|
|
|
|
Timer? _reconnectTimer;
|
|
Duration _currentReconnectDelay = Duration.zero;
|
|
|
|
String? _lastToken;
|
|
|
|
RemoteAddedQueue? _addedQueue;
|
|
|
|
/// Imposta la queue per gestire gli eventi "added" internamente.
|
|
void setAddedQueue(RemoteAddedQueue queue) {
|
|
_addedQueue = queue;
|
|
debugPrint('[remote][ws] addedQueue set');
|
|
}
|
|
|
|
// Last seen persistito in RemoteStateStore.
|
|
int _lastSeenMs = 0;
|
|
|
|
// Stato handshake.
|
|
bool _authed = false;
|
|
|
|
// Il server ha richiesto recovery e aspetta recovery_done.
|
|
bool _needRecoveryDoneAck = false;
|
|
|
|
// Recovery richiesta localmente perché la WS era dormiente.
|
|
bool _recoveryAfterDormant = false;
|
|
|
|
// Debounce recovery.
|
|
Timer? _recoveryDebounce;
|
|
bool _recoveryScheduled = false;
|
|
|
|
bool get isConnected => _ch != null && _authed;
|
|
|
|
Future<void> connect({required String token}) async {
|
|
_lastToken = token;
|
|
|
|
if (_connecting) return;
|
|
if (isConnected) return;
|
|
|
|
_connecting = true;
|
|
_closing = false;
|
|
|
|
try {
|
|
// Chiudi eventuale connessione precedente.
|
|
await close();
|
|
|
|
await deduper.init();
|
|
|
|
final sessionId = await store.getOrCreateSessionId();
|
|
|
|
// Carica lastSeen persistito.
|
|
_lastSeenMs = await store.getWsLastSeenMs();
|
|
|
|
final now = DateTime.now().millisecondsSinceEpoch;
|
|
final dormant = _lastSeenMs != 0 &&
|
|
(now - _lastSeenMs) > dormantWindow.inMilliseconds;
|
|
|
|
if (dormant) {
|
|
debugPrint(
|
|
'[remote][ws] dormant (> ${dormantWindow.inSeconds}s) before connect '
|
|
'-> keep session_id and schedule progressive recovery',
|
|
);
|
|
|
|
_recoveryAfterDormant = true;
|
|
}
|
|
|
|
// Mantieni sempre la sessione esistente.
|
|
// Non fare reset session_id per dormancy, altrimenti il server
|
|
// potrebbe interpretarla come sessione nuova.
|
|
final finalSessionId = sessionId;
|
|
|
|
final uri = Uri.parse(wsUrl);
|
|
|
|
_ch = IOWebSocketChannel.connect(
|
|
uri,
|
|
pingInterval: pingInterval,
|
|
);
|
|
|
|
_authed = false;
|
|
_needRecoveryDoneAck = false;
|
|
|
|
_currentReconnectDelay = reconnectDelay;
|
|
|
|
_sub = _ch!.stream.listen(
|
|
(raw) async {
|
|
try {
|
|
final rawStr = raw is String ? raw : raw.toString();
|
|
debugPrint('[remote][ws][raw] $rawStr');
|
|
|
|
final decodedForLog = _decode(raw);
|
|
if (decodedForLog != null) {
|
|
final copy = Map<String, dynamic>.from(decodedForLog);
|
|
|
|
if (copy.containsKey('token')) {
|
|
copy['token'] = '<redacted>';
|
|
}
|
|
|
|
if (copy.containsKey('auth') && copy['auth'] is Map) {
|
|
final authMap = Map<String, dynamic>.from(copy['auth'] as Map);
|
|
if (authMap.containsKey('token')) {
|
|
authMap['token'] = '<redacted>';
|
|
}
|
|
copy['auth'] = authMap;
|
|
}
|
|
|
|
debugPrint(
|
|
'[remote][ws][decoded]\n'
|
|
'${const JsonEncoder.withIndent(' ').convert(copy)}',
|
|
);
|
|
} else {
|
|
debugPrint('[remote][ws][decoded] null (parse failed)');
|
|
}
|
|
} catch (e, st) {
|
|
debugPrint('[remote][ws][debug-print] error=$e\n$st');
|
|
}
|
|
|
|
_touchLastSeen();
|
|
|
|
final msg = _decode(raw);
|
|
if (msg == null) return;
|
|
|
|
final type = msg['type']?.toString();
|
|
|
|
try {
|
|
final eventId = msg['event_id']?.toString();
|
|
final id = msg['id']?.toString();
|
|
debugPrint('[remote][ws][evt] type=$type event_id=$eventId id=$id');
|
|
} catch (_) {
|
|
// ignore
|
|
}
|
|
|
|
// auth_error dal server.
|
|
if (type == 'auth_error') {
|
|
debugPrint('[remote][ws] auth_error=${msg['error']}');
|
|
|
|
onDisconnected?.call();
|
|
|
|
_scheduleRecovery();
|
|
await close();
|
|
return;
|
|
}
|
|
|
|
// auth_ok: il server può chiedere una recovery progressiva.
|
|
if (type == 'auth_ok') {
|
|
_authed = true;
|
|
|
|
final needFull = msg['need_full_sync'] == true;
|
|
final hasNeedRecoveryField = msg.containsKey('need_recovery');
|
|
final needRecoveryFlag = msg['need_recovery'] == true;
|
|
|
|
// Se il server manda esplicitamente need_recovery,
|
|
// il client rispetta quella decisione.
|
|
//
|
|
// _recoveryAfterDormant resta solo fallback per server vecchi
|
|
// che non mandano ancora need_recovery.
|
|
final needRecovery = needFull ||
|
|
needRecoveryFlag ||
|
|
(!hasNeedRecoveryField && _recoveryAfterDormant);
|
|
|
|
debugPrint(
|
|
'[remote][ws] auth_ok '
|
|
'user=${msg['user']} '
|
|
'session=${msg['session_id']} '
|
|
'need_full_sync=$needFull '
|
|
'need_recovery=$needRecoveryFlag '
|
|
'hasNeedRecoveryField=$hasNeedRecoveryField '
|
|
'recoveryAfterDormant=$_recoveryAfterDormant '
|
|
'needRecovery=$needRecovery '
|
|
'reason=${msg['reason']}',
|
|
);
|
|
|
|
if (needRecovery) {
|
|
_needRecoveryDoneAck = true;
|
|
|
|
// Se il server chiede full sync legacy, mantieni il delay.
|
|
// Se è solo need_recovery/dormant, parti subito.
|
|
_scheduleRecovery(
|
|
delay: needFull ? needFullSyncDelay : Duration.zero,
|
|
);
|
|
} else {
|
|
// Nessuna recovery necessaria: avvisa il server che siamo pronti.
|
|
_send({'type': 'client_recovered'});
|
|
|
|
// WS autenticato e nessuna recovery richiesta:
|
|
// ora RemoteController può mettere RemoteSyncState.upToDate.
|
|
onReady?.call();
|
|
}
|
|
|
|
_recoveryAfterDormant = false;
|
|
return;
|
|
}
|
|
|
|
// Il server può chiedere recovery anche dopo auth_ok,
|
|
// ad esempio dopo client_recovered o se vede sessione dormiente.
|
|
if (type == 'need_recovery') {
|
|
debugPrint(
|
|
'[remote][ws] need_recovery '
|
|
'reason=${msg['reason']}',
|
|
);
|
|
|
|
_needRecoveryDoneAck = true;
|
|
_scheduleRecovery(delay: Duration.zero);
|
|
return;
|
|
}
|
|
|
|
// ping/pong.
|
|
if (type == 'ping') {
|
|
_send({'type': 'pong'});
|
|
return;
|
|
}
|
|
|
|
final eventId = msg['event_id']?.toString();
|
|
|
|
// Dedupe -> ACK e stop.
|
|
if (eventId != null && deduper.has(eventId)) {
|
|
_send({'type': 'ack', 'event_id': eventId});
|
|
return;
|
|
}
|
|
|
|
// Gestione rapida "added" tramite queue se presente.
|
|
final msgType = msg['type']?.toString();
|
|
if (msgType == 'added') {
|
|
final id = msg['id']?.toString();
|
|
|
|
if (_addedQueue != null) {
|
|
try {
|
|
_addedQueue!.enqueue(id);
|
|
|
|
if (eventId != null) {
|
|
await deduper.mark(eventId);
|
|
_send({'type': 'ack', 'event_id': eventId});
|
|
}
|
|
|
|
return;
|
|
} catch (e, st) {
|
|
debugPrint('[remote][ws] addedQueue error=$e\n$st');
|
|
// fallthrough: delegate to onEvent as fallback
|
|
}
|
|
}
|
|
}
|
|
|
|
// Gestione evento: ACK solo se handled=true.
|
|
bool handled = false;
|
|
|
|
try {
|
|
handled = await onEvent(msg);
|
|
} catch (e, st) {
|
|
debugPrint('[remote][ws] onEvent error=$e\n$st');
|
|
|
|
handled = false;
|
|
|
|
// Se fallisce, chiedi recovery ma NON ACK.
|
|
_scheduleRecovery();
|
|
}
|
|
|
|
if (eventId != null && handled) {
|
|
await deduper.mark(eventId);
|
|
_send({'type': 'ack', 'event_id': eventId});
|
|
} else {
|
|
// NO ACK: server ritenterà.
|
|
}
|
|
},
|
|
onDone: () {
|
|
debugPrint('[remote][ws] stream closed');
|
|
|
|
final wasAuthed = _authed;
|
|
|
|
_cleanupChannelOnly();
|
|
|
|
if (!_closing) {
|
|
onDisconnected?.call();
|
|
|
|
if (wasAuthed) {
|
|
_scheduleRecovery();
|
|
}
|
|
|
|
_scheduleReconnect();
|
|
}
|
|
},
|
|
onError: (e) {
|
|
debugPrint('[remote][ws] stream error=$e');
|
|
|
|
final wasAuthed = _authed;
|
|
|
|
_cleanupChannelOnly();
|
|
|
|
if (!_closing) {
|
|
onDisconnected?.call();
|
|
|
|
if (wasAuthed) {
|
|
_scheduleRecovery();
|
|
}
|
|
|
|
_scheduleReconnect();
|
|
}
|
|
},
|
|
cancelOnError: false,
|
|
);
|
|
|
|
// Auth handshake.
|
|
_send({
|
|
'type': 'auth',
|
|
'token': token,
|
|
'session_id': finalSessionId,
|
|
});
|
|
} finally {
|
|
_connecting = false;
|
|
}
|
|
}
|
|
|
|
Future<void> close() async {
|
|
_closing = true;
|
|
|
|
_reconnectTimer?.cancel();
|
|
_reconnectTimer = null;
|
|
|
|
_recoveryDebounce?.cancel();
|
|
_recoveryDebounce = null;
|
|
_recoveryScheduled = false;
|
|
|
|
try {
|
|
await _sub?.cancel();
|
|
_sub = null;
|
|
|
|
final ch = _ch;
|
|
_ch = null;
|
|
|
|
_authed = false;
|
|
|
|
if (ch != null) {
|
|
await ch.sink.close();
|
|
}
|
|
} catch (_) {
|
|
// ignore
|
|
} finally {
|
|
_cleanupAll();
|
|
}
|
|
}
|
|
|
|
/// Da chiamare quando hai completato una recovery richiesta dal server
|
|
/// tramite need_full_sync=true, need_recovery=true o recovery dopo dormancy.
|
|
/// Invia recovery_done una sola volta per sessione.
|
|
void notifyRecoveryCompleted() {
|
|
if (!_needRecoveryDoneAck) return;
|
|
if (!isConnected) return;
|
|
|
|
_needRecoveryDoneAck = false;
|
|
|
|
_send({'type': 'recovery_done'});
|
|
|
|
// Recovery completata: ora RemoteController può mostrare i remoti.
|
|
onReady?.call();
|
|
}
|
|
|
|
// --------------------------
|
|
// reconnect logic
|
|
// --------------------------
|
|
|
|
void _scheduleReconnect() {
|
|
if (!autoReconnect) return;
|
|
|
|
if (isConnected) return;
|
|
if (_connecting) return;
|
|
if (_reconnectTimer != null) return;
|
|
|
|
final token = _lastToken;
|
|
if (token == null || token.isEmpty) return;
|
|
|
|
final delay = _currentReconnectDelay == Duration.zero
|
|
? reconnectDelay
|
|
: _currentReconnectDelay;
|
|
|
|
_currentReconnectDelay = _nextDelay(delay);
|
|
|
|
debugPrint('[remote][ws] scheduling reconnect in ${delay.inMilliseconds}ms');
|
|
|
|
_reconnectTimer = Timer(delay, () async {
|
|
_reconnectTimer = null;
|
|
|
|
if (_closing) return;
|
|
if (isConnected) return;
|
|
if (_connecting) return;
|
|
|
|
try {
|
|
await connect(token: token);
|
|
} catch (e, st) {
|
|
debugPrint('[remote][ws] reconnect failed: $e\n$st');
|
|
|
|
_cleanupChannelOnly();
|
|
onDisconnected?.call();
|
|
|
|
_scheduleReconnect();
|
|
}
|
|
});
|
|
}
|
|
|
|
Duration _nextDelay(Duration current) {
|
|
final nextMs = (current.inMilliseconds * 2).clamp(
|
|
reconnectDelay.inMilliseconds,
|
|
maxReconnectDelay.inMilliseconds,
|
|
);
|
|
|
|
return Duration(milliseconds: nextMs);
|
|
}
|
|
|
|
// --------------------------
|
|
// recovery debounce
|
|
// --------------------------
|
|
|
|
void _scheduleRecovery({
|
|
Duration delay = const Duration(milliseconds: 250),
|
|
}) {
|
|
if (_recoveryScheduled) return;
|
|
|
|
_recoveryScheduled = true;
|
|
|
|
_recoveryDebounce?.cancel();
|
|
_recoveryDebounce = Timer(delay, () async {
|
|
try {
|
|
await onNeedRecoverySync();
|
|
} catch (e, st) {
|
|
debugPrint('[remote][ws] onNeedRecoverySync error=$e\n$st');
|
|
} finally {
|
|
_recoveryScheduled = false;
|
|
}
|
|
});
|
|
}
|
|
|
|
// --------------------------
|
|
// utils
|
|
// --------------------------
|
|
|
|
void _touchLastSeen() {
|
|
_lastSeenMs = DateTime.now().millisecondsSinceEpoch;
|
|
unawaited(store.setWsLastSeenMs(_lastSeenMs));
|
|
}
|
|
|
|
void _send(Map<String, dynamic> obj) {
|
|
try {
|
|
final payload = jsonEncode(obj);
|
|
|
|
_ch?.sink.add(payload);
|
|
|
|
debugPrint('[remote][ws] send: ${obj['type'] ?? '<unknown>'}');
|
|
} catch (e, st) {
|
|
debugPrint('[remote][ws] send failed: $e\n$st');
|
|
}
|
|
}
|
|
|
|
Map<String, dynamic>? _decode(dynamic raw) {
|
|
try {
|
|
final s = raw is String ? raw : raw.toString();
|
|
final v = jsonDecode(s);
|
|
|
|
return v is Map<String, dynamic> ? v : null;
|
|
} catch (_) {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
void _cleanupChannelOnly() {
|
|
_sub?.cancel();
|
|
_sub = null;
|
|
|
|
_ch = null;
|
|
|
|
_authed = false;
|
|
_connecting = false;
|
|
}
|
|
|
|
void _cleanupAll() {
|
|
_cleanupChannelOnly();
|
|
_closing = false;
|
|
}
|
|
}
|