comunes/tane
1
0
Fork 0
Code Issues Pull requests Projects Releases 7 Packages Wiki Activity Actions

refactor(social): one shared relay connection per identity

Browse source 
Every feature (offers, messaging, trust, profile, the inbox listener) used
to open its OWN RelayPool via social.openSession — several sockets to the
same relays, more battery, and reconnection logic living only in the inbox
listener.

Add SocialConnection: ONE shared session per identity, lazily connected and
reused by all. It watches connectivity — dropping the session when the
network goes and reconnecting when it returns, announcing each change on a
sessions stream so the inbox listener re-subscribes automatically. Callers
get the session via connection.session() and never close it (the connection
owns its lifecycle); recreated/disposed on an identity switch.

- InboxService now consumes the shared connection (subscribes on its
  sessions stream) instead of owning its own socket + connectivity code.
- createOffersCubit + the chat/market/profile screens use the shared
  connection; dead createMessagesCubit/createTrustCubit factories removed.
- DI registers SocialConnection per identity; Bootstrap starts it (after
  the inbox subscribes); switchSocialAccount disposes + recreates it.

Tests: SocialConnection lifecycle (reuse, concurrent connect, offline drop +
reconnect, unreachable retry) with a fake opener + online stream; inbox test
updated. nostr added as a dev_dependency for the channel fake.
This commit is contained in:
vjrj 2026-07-10 21:41:17 +02:00
parent e2b88b4f26
commit bb4ee2fd89
19 changed files with 431 additions and 280 deletions
Download patch file Download diff file Expand all files Collapse all files

120
apps/app_seeds/lib/services/inbox_service.dart
View file

@ -1,110 +1,89 @@
import 'dart:async';
import 'package:commons_core/commons_core.dart';
import 'package:connectivity_plus/connectivity_plus.dart';
import 'package:flutter/foundation.dart';
import '../i18n/strings.g.dart';
import 'message_store.dart';
import 'notification_service.dart';
import 'profile_cache.dart';
import 'social_service.dart';
import 'social_settings.dart';
import 'social_connection.dart';
import 'social_service.dart' show SocialSession;
import 'unread_service.dart';
/// App-wide inbox listener for private messages (NIP-17).
///
/// Without this, a message only arrived while its specific chat screen was
/// open — so a first message from a new peer was invisible (nothing in local
/// storage, so nothing in the inbox list, so you never opened the chat, so you
/// never subscribed). This keeps ONE long-lived subscription for the whole app
/// while online: it persists every incoming message to the [MessageStore] and
/// fires [changes] so the inbox list refreshes. Reconnects when the network
/// returns; degrades to nothing offline (local-first).
/// Without this, a message only arrived while its specific chat screen was open
/// — so a first message from a new peer was invisible. This keeps one inbox
/// subscription over the SHARED [SocialConnection] for the whole app while
/// online: it persists every incoming message to the [MessageStore], updates
/// unread counts, fires an OS notification, and fires [changes] so the inbox
/// list refreshes. It re-subscribes automatically each time the connection
/// reconnects (and stops when it drops), because it listens to the connection's
/// session stream rather than owning a socket itself.
///
/// Foreground only — background/push delivery is a later concern.
class InboxService {
InboxService({
required SocialService social,
required SocialSettings settings,
required SocialConnection connection,
required String selfPubkey,
required MessageStore store,
ProfileCache? profileCache,
UnreadService? unread,
NotificationService? notifications,
}) : _social = social,
_settings = settings,
}) : _connection = connection,
_selfPubkey = selfPubkey,
_store = store,
_profileCache = profileCache,
_unread = unread,
_notifications = notifications;
final SocialService _social;
final SocialSettings _settings;
final SocialConnection _connection;
final String _selfPubkey;
final MessageStore _store;
final ProfileCache? _profileCache;
final UnreadService? _unread;
final NotificationService? _notifications;
final _changes = StreamController<void>.broadcast();
SocialSession? _session;
StreamSubscription<SocialSession?>? _sessionsSub;
StreamSubscription<PrivateMessage>? _inboxSub;
StreamSubscription<List<ConnectivityResult>>? _connSub;
SocialSession? _session;
bool _started = false;
bool _connecting = false;
/// Fires (with no payload) after a NEW message is persisted — the inbox list
/// listens to this to reload. Broadcast, so several screens can listen.
/// Fires (no payload) after a NEW message is persisted — the inbox list
/// listens to reload. Broadcast, so several screens can listen.
Stream<void> get changes => _changes.stream;
/// Begins listening. Idempotent. Connects now if online, and (re)connects
/// whenever the network returns. Safe to call once at app start.
Future<void> start() async {
/// Begins listening. Subscribe to the connection's sessions BEFORE the
/// connection starts connecting, so the first session is caught too.
void start() {
if (_started) return;
_started = true;
try {
_connSub = Connectivity().onConnectivityChanged.listen((results) {
final offline =
results.isEmpty || results.every((r) => r == ConnectivityResult.none);
if (offline) {
_dropSession();
} else {
unawaited(_connect());
}
});
} catch (_) {
// Platform without connectivity support — just try once below.
}
await _connect();
_sessionsSub = _connection.sessions.listen(_onSession);
// In case the connection is already up (a screen connected first).
final current = _connection.current;
if (current != null) _onSession(current);
}
/// Opens a session and subscribes to the inbox, if not already connected.
/// Guarded so overlapping connectivity events can't open two sessions.
Future<void> _connect() async {
if (_session != null || _connecting) return; // already (being) connected
_connecting = true;
try {
final relays = await _settings.relayUrls();
if (relays.isEmpty) return; // offline / unconfigured
final session = await _social.openSession(relays);
if (!_started) {
await session.close(); // stopped while connecting
return;
}
_session = session;
/// (Re)binds the inbox subscription to [session] — the connection handing us a
/// fresh session (reconnect) or null (dropped).
void _onSession(SocialSession? session) {
if (identical(session, _session) && _inboxSub != null) return;
unawaited(_inboxSub?.cancel());
_inboxSub = null;
_session = session;
if (session != null) {
_inboxSub = session.messages.inbox().listen(
ingest,
onError: (_) => _dropSession(), // relay dropped — retry on next change
);
} catch (_) {
// No relay reachable — a later connectivity change retries.
} finally {
_connecting = false;
ingest,
onError: (_) {}, // drop handled by the connection's reconnect
);
}
}
/// Persists one incoming [message] (deduped) and, when it's new, fires
/// [changes] and best-effort caches the sender's display name. Separated from
/// the transport so it can be unit-tested without a relay.
/// Persists one incoming [message] (deduped) and, when new, fires [changes],
/// updates unread, notifies, and best-effort caches the sender's name. A
/// testable seam (no relay).
@visibleForTesting
Future<void> ingest(PrivateMessage message) async {
final stored = await _store.append(message.fromPubkey, message);
@ -113,9 +92,8 @@ class InboxService {
final peer = message.fromPubkey;
// Defensive: never badge or notify for your own message. NIP-17 doesn't
// loop a sent gift wrap back to its author, so this shouldn't happen — the
// guard just keeps the seam honest.
if (peer != _social.publicKeyHex) {
// loop a sent gift wrap back to its author; the guard keeps the seam honest.
if (peer != _selfPubkey) {
await _unread?.onMessageReceived(peer, message);
await _maybeNotify(peer);
}
@ -151,20 +129,12 @@ class InboxService {
}
}
void _dropSession() {
unawaited(_inboxSub?.cancel());
_inboxSub = null;
unawaited(_session?.close());
_session = null;
}
Future<void> stop() async {
_started = false;
await _connSub?.cancel();
_connSub = null;
await _sessionsSub?.cancel();
_sessionsSub = null;
await _inboxSub?.cancel();
_inboxSub = null;
await _session?.close();
_session = null;
if (!_changes.isClosed) await _changes.close();
}

107
apps/app_seeds/lib/services/social_connection.dart Normal file
View file

@ -0,0 +1,107 @@
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<SocialSession> Function(List<String> 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<bool>? online,
}) : _settings = settings,
_open = open ?? social.openSession,
_online = online;
final SocialSettings _settings;
final SessionOpener _open;
final Stream<bool>? _online;
final _sessions = StreamController<SocialSession?>.broadcast();
SocialSession? _current;
Future<SocialSession?>? _pending;
StreamSubscription<bool>? _onlineSub;
bool _disposed = false;
/// Emits the live session on each (re)connect, and null when it drops.
Stream<SocialSession?> 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. Idempotent-ish; call once at startup.
void start() {
_onlineSub = (_online ?? _connectivityOnline()).listen((isOnline) {
if (!isOnline) {
_drop();
} else if (_current == null) {
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<SocialSession?> session() {
if (_current != null) return Future.value(_current);
return _pending ??= _connect();
}
Future<SocialSession?> _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;
_sessions.add(s);
return s;
} catch (_) {
return null; // unreachable — a later connectivity change retries
} finally {
_pending = null;
}
}
void _drop() {
final s = _current;
_current = null;
if (s != null) {
unawaited(s.close());
if (!_sessions.isClosed) _sessions.add(null);
}
}
Future<void> dispose() async {
_disposed = true;
await _onlineSub?.cancel();
_onlineSub = null;
_drop();
if (!_sessions.isClosed) await _sessions.close();
}
static Stream<bool> _connectivityOnline() =>
Connectivity().onConnectivityChanged.map((results) =>
results.any((r) => r != ConnectivityResult.none));
}