import 'dart:async'; import 'package:connectivity_plus/connectivity_plus.dart'; import 'social_service.dart'; import 'social_settings.dart'; /// Opens a session against the given relays. Injectable so the connection logic /// can be unit-tested without a real relay. typedef SessionOpener = Future Function(List relays); /// ONE shared relay connection per identity, reused by every social feature /// (offers, messaging, trust, profile, the inbox listener) instead of each /// opening its own `RelayPool`. Fewer sockets, less battery, one place to manage /// reconnection. /// /// Lazily connects on first [session] call; reconnects when the network returns /// and drops the session when it goes away, announcing each change on [sessions] /// so long-lived consumers (the inbox listener) can re-subscribe. Callers must /// NOT close the session they get — this owns its lifecycle. Recreated on an /// identity switch (the old one is disposed). class SocialConnection { SocialConnection({ required SocialService social, required SocialSettings settings, SessionOpener? open, Stream? online, List? retrySchedule, }) : _settings = settings, _open = open ?? social.openSession, _online = online, _retrySchedule = retrySchedule ?? _defaultRetrySchedule; /// Backoff for self-retries after a failed attempt while started and not /// knowingly offline (the fresh-install case: online the whole time, first /// connect fails, so no connectivity change ever retriggers a connect). static const _defaultRetrySchedule = [ Duration(seconds: 5), Duration(seconds: 15), Duration(seconds: 45), Duration(seconds: 90), ]; final SocialSettings _settings; final SessionOpener _open; final Stream? _online; final List _retrySchedule; final _sessions = StreamController.broadcast(); SocialSession? _current; Future? _pending; StreamSubscription? _onlineSub; bool _disposed = false; bool _started = false; bool _knownOffline = false; Timer? _retryTimer; int _retryIndex = 0; /// Emits the live session on each (re)connect, and null when it drops. Stream get sessions => _sessions.stream; /// The current shared session, or null if not connected right now. SocialSession? get current => _current; /// Begins watching connectivity (reconnect on regain, drop when offline) and /// attempts an initial connect. Called at startup when sharing is already on, /// and again the moment someone joins the sharing side — hence the guard, so /// a second call never stacks a second connectivity subscription. void start() { if (_started || _disposed) return; _started = true; _onlineSub = (_online ?? _connectivityOnline()).listen((isOnline) { _knownOffline = !isOnline; if (!isOnline) { _cancelRetry(); _drop(); } else if (_current == null) { _retryIndex = 0; // fresh network — start the backoff over unawaited(session()); } }); unawaited(session()); // initial attempt (also connects if already online) } /// The shared session, connecting on first use. Returns null when offline or /// no relay is reachable; the feature then degrades gracefully. Future session() { if (_current != null) return Future.value(_current); return _pending ??= _connect(); } Future _connect() async { try { final relays = await _settings.relayUrls(); if (relays.isEmpty) return null; final s = await _open(relays); if (_disposed) { await s.close(); return null; } _current = s; _retryIndex = 0; _cancelRetry(); _sessions.add(s); return s; } catch (_) { _scheduleRetry(); // unreachable — retry with backoff (see below) return null; } finally { _pending = null; } } /// After a failed attempt, retries by itself while started and not knowingly /// offline — a connectivity change may never come if the device was online /// all along. Un-started connections (one-shot [session] callers, tests) /// never leave a timer behind. void _scheduleRetry() { if (!_started || _disposed || _knownOffline || _current != null) return; _cancelRetry(); final i = _retryIndex < _retrySchedule.length ? _retryIndex : _retrySchedule.length - 1; _retryIndex++; _retryTimer = Timer(_retrySchedule[i], () { if (_disposed || _current != null) return; unawaited(session()); }); } void _cancelRetry() { _retryTimer?.cancel(); _retryTimer = null; } void _drop() { final s = _current; _current = null; if (s != null) { unawaited(s.close()); if (!_sessions.isClosed) _sessions.add(null); } } /// Goes offline for good until [start] is called again: stops watching /// connectivity, cancels any pending retry and tears the live session down. /// This is what "turn sharing off" must do — before it existed, clearing the /// server list only took effect on the next launch, because the already-open /// session was never closed. Unlike [dispose] the object stays usable. Future stop() async { _started = false; _cancelRetry(); await _onlineSub?.cancel(); _onlineSub = null; _drop(); } Future dispose() async { _disposed = true; _cancelRetry(); await _onlineSub?.cancel(); _onlineSub = null; _drop(); if (!_sessions.isClosed) await _sessions.close(); } static Stream _connectivityOnline() => Connectivity().onConnectivityChanged.map((results) => results.any((r) => r != ConnectivityResult.none)); }