diff --git a/apps/app_seeds/lib/services/social_service.dart b/apps/app_seeds/lib/services/social_service.dart index a2b1b43..aee6ea6 100644 --- a/apps/app_seeds/lib/services/social_service.dart +++ b/apps/app_seeds/lib/services/social_service.dart @@ -36,30 +36,30 @@ class SocialService { return SocialService(identity: identity, relays: relays); } - /// Opens a session against [relayUrl]: one shared connection carrying all - /// three transports (offers, messaging, trust). Caller closes it. - Future openSession(String relayUrl) async { - final connection = - await NostrConnection.connect(relayUrl, identity: identity); - return SocialSession(connection); + /// Opens a session against [relayUrls]: one fault-tolerant pool carrying all + /// three transports (offers, messaging, trust). Relays that are down are + /// skipped; throws only when none are reachable. Caller closes it. + Future openSession(List relayUrls) async { + final pool = await RelayPool.connect(relayUrls, identity: identity); + return SocialSession(pool); } } -/// One live relay connection with the three transports on top — the -/// "one connection, three interfaces" shape, at the app layer. +/// One relay channel with the three transports on top — the +/// "one channel, three interfaces" shape, at the app layer. class SocialSession { - SocialSession(this._connection) - : offers = NostrOfferTransport(_connection), - messages = NostrMessageTransport(_connection), - trust = NostrTrustTransport(_connection); + SocialSession(this._channel) + : offers = NostrOfferTransport(_channel), + messages = NostrMessageTransport(_channel), + trust = NostrTrustTransport(_channel); - final NostrConnection _connection; + final NostrChannel _channel; final OfferTransport offers; final MessageTransport messages; final TrustTransport trust; - Future close() => _connection.close(); + Future close() => _channel.close(); } Uint8List _hexToBytes(String hex) { diff --git a/apps/app_seeds/lib/state/offers_cubit.dart b/apps/app_seeds/lib/state/offers_cubit.dart index 667fee0..42acc4b 100644 --- a/apps/app_seeds/lib/state/offers_cubit.dart +++ b/apps/app_seeds/lib/state/offers_cubit.dart @@ -172,9 +172,9 @@ Future createOffersCubit( final relays = await settings.relayUrls(); if (relays.isEmpty) return OffersCubit(null); try { - final session = await social.openSession(relays.first); + final session = await social.openSession(relays); return OffersCubit(session.offers, onDispose: session.close); } catch (_) { - return OffersCubit(null); // offline / relay unreachable + return OffersCubit(null); // offline / no relay reachable } } diff --git a/packages/commons_core/lib/commons_core.dart b/packages/commons_core/lib/commons_core.dart index 8fca064..44fc753 100644 --- a/packages/commons_core/lib/commons_core.dart +++ b/packages/commons_core/lib/commons_core.dart @@ -14,10 +14,12 @@ export 'src/ids/id_gen.dart'; export 'src/social/certification.dart'; export 'src/social/geohash.dart'; export 'src/social/message_transport.dart'; +export 'src/social/nostr/nostr_channel.dart'; export 'src/social/nostr/nostr_connection.dart'; export 'src/social/nostr/nostr_message_transport.dart'; export 'src/social/nostr/nostr_offer_transport.dart'; export 'src/social/nostr/nostr_trust_transport.dart'; +export 'src/social/nostr/relay_pool.dart'; export 'src/social/offer.dart'; export 'src/social/offer_transport.dart'; export 'src/social/trust_transport.dart'; diff --git a/packages/commons_core/lib/src/social/nostr/nostr_channel.dart b/packages/commons_core/lib/src/social/nostr/nostr_channel.dart new file mode 100644 index 0000000..ceecfb8 --- /dev/null +++ b/packages/commons_core/lib/src/social/nostr/nostr_channel.dart @@ -0,0 +1,21 @@ +import 'package:nostr/nostr.dart'; + +/// What the three transports need from "the network": sign with an identity, +/// publish an event, and run subscriptions. Implemented by a single +/// [NostrConnection] and by a fault-tolerant [RelayPool] over several relays, so +/// the transports don't care whether they talk to one relay or many. +abstract interface class NostrChannel { + String get privateKeyHex; + String get publicKeyHex; + + /// Publishes [event]; accepted if at least one relay accepts it. + Future<({bool accepted, String message})> publish(Event event); + + /// Streams events matching [filter] from every relay, de-duplicated by id. + Stream subscribe(Filter filter); + + /// Collects stored events matching [filter] up to EOSE, de-duplicated by id. + Future> reqOnce(Filter filter); + + Future close(); +} diff --git a/packages/commons_core/lib/src/social/nostr/nostr_connection.dart b/packages/commons_core/lib/src/social/nostr/nostr_connection.dart index 0a0abf3..bcfd345 100644 --- a/packages/commons_core/lib/src/social/nostr/nostr_connection.dart +++ b/packages/commons_core/lib/src/social/nostr/nostr_connection.dart @@ -5,6 +5,7 @@ import 'dart:io'; import 'package:nostr/nostr.dart'; import '../../identity/nostr_key_derivation.dart'; +import 'nostr_channel.dart'; /// The ONE piece shared by every Nostr-backed transport: a single relay socket /// plus the signing identity and the EVENT/OK, REQ/EOSE lifecycle. @@ -14,7 +15,7 @@ import '../../identity/nostr_key_derivation.dart'; /// Pure Dart (`dart:io` WebSocket) — no Flutter. A production relay *pool* /// (multiple relays, reconnection, backoff) would wrap several of these; the /// spike/foundation uses one relay. -class NostrConnection { +class NostrConnection implements NostrChannel { NostrConnection._(this._socket, this.identity) { _socket.listen(_onData, onDone: _onDone, onError: (_) => _onDone()); } @@ -24,7 +25,9 @@ class NostrConnection { /// The derived identity this connection signs with. final NostrIdentity identity; + @override String get publicKeyHex => identity.publicKeyHex; + @override String get privateKeyHex => identity.privateKeyHex; final _incoming = StreamController>.broadcast(); @@ -47,6 +50,7 @@ class NostrConnection { } /// Publishes an already-signed [event]; resolves with the relay's OK verdict. + @override Future<({bool accepted, String message})> publish(Event event) async { final ok = _incoming.stream.firstWhere( (m) => m[0] == 'OK' && m[1] == event.id, @@ -57,6 +61,7 @@ class NostrConnection { } /// Streams events matching [filter] (live), until the caller cancels. + @override Stream subscribe(Filter filter) { final subId = 'sub${_subCounter++}'; late StreamController controller; @@ -81,6 +86,7 @@ class NostrConnection { } /// Collects stored events matching [filter] up to EOSE, then closes the sub. + @override Future> reqOnce(Filter filter) async { final subId = 'once${_subCounter++}'; final results = []; @@ -100,6 +106,7 @@ class NostrConnection { return results; } + @override Future close() async { await _socket.close(); if (!_incoming.isClosed) await _incoming.close(); diff --git a/packages/commons_core/lib/src/social/nostr/nostr_message_transport.dart b/packages/commons_core/lib/src/social/nostr/nostr_message_transport.dart index c68d1c1..2d5389a 100644 --- a/packages/commons_core/lib/src/social/nostr/nostr_message_transport.dart +++ b/packages/commons_core/lib/src/social/nostr/nostr_message_transport.dart @@ -3,7 +3,7 @@ import 'dart:async'; import 'package:nostr/nostr.dart' as nostr; import '../message_transport.dart'; -import 'nostr_connection.dart'; +import 'nostr_channel.dart'; /// NIP-17 private-DM backend for [MessageTransport], on the shared /// [NostrConnection]. The gift-wrap onion (rumor kind 14 → seal kind 13 → wrap @@ -12,7 +12,7 @@ import 'nostr_connection.dart'; class NostrMessageTransport implements MessageTransport { NostrMessageTransport(this._conn); - final NostrConnection _conn; + final NostrChannel _conn; static const kindGiftWrap = 1059; diff --git a/packages/commons_core/lib/src/social/nostr/nostr_offer_transport.dart b/packages/commons_core/lib/src/social/nostr/nostr_offer_transport.dart index f17c84c..c9408b9 100644 --- a/packages/commons_core/lib/src/social/nostr/nostr_offer_transport.dart +++ b/packages/commons_core/lib/src/social/nostr/nostr_offer_transport.dart @@ -3,13 +3,14 @@ import 'package:nostr/nostr.dart'; import '../offer.dart'; import '../offer_transport.dart'; import 'nip99_codec.dart'; -import 'nostr_connection.dart'; +import 'nostr_channel.dart'; -/// Nostr NIP-99 backend for [OfferTransport], on a shared [NostrConnection]. +/// Nostr NIP-99 backend for [OfferTransport], on a shared channel (one relay or +/// a [RelayPool]). class NostrOfferTransport implements OfferTransport { NostrOfferTransport(this._conn); - final NostrConnection _conn; + final NostrChannel _conn; final Nip99Codec _codec = Nip99Codec(); Filter _filter(DiscoveryQuery q) => Filter( diff --git a/packages/commons_core/lib/src/social/nostr/nostr_trust_transport.dart b/packages/commons_core/lib/src/social/nostr/nostr_trust_transport.dart index 03565b6..bbde9b8 100644 --- a/packages/commons_core/lib/src/social/nostr/nostr_trust_transport.dart +++ b/packages/commons_core/lib/src/social/nostr/nostr_trust_transport.dart @@ -2,7 +2,7 @@ import 'package:nostr/nostr.dart'; import '../certification.dart'; import '../trust_transport.dart'; -import 'nostr_connection.dart'; +import 'nostr_channel.dart'; /// Duniter-style certifications carried as Nostr events, on the shared /// [NostrConnection]. No settled NIP exists for a web of trust, so this uses a @@ -12,7 +12,7 @@ import 'nostr_connection.dart'; class NostrTrustTransport implements TrustTransport { NostrTrustTransport(this._conn); - final NostrConnection _conn; + final NostrChannel _conn; /// Custom Tanemaki certification kind (addressable range). static const kindCertification = 30777; diff --git a/packages/commons_core/lib/src/social/nostr/relay_pool.dart b/packages/commons_core/lib/src/social/nostr/relay_pool.dart new file mode 100644 index 0000000..eac038d --- /dev/null +++ b/packages/commons_core/lib/src/social/nostr/relay_pool.dart @@ -0,0 +1,123 @@ +import 'dart:async'; + +import 'package:nostr/nostr.dart'; + +import '../../identity/nostr_key_derivation.dart'; +import 'nostr_channel.dart'; +import 'nostr_connection.dart'; + +/// A fault-tolerant [NostrChannel] over several relays — the relay-strategy +/// hardening (network-trust.md §3: a few community relays for reliability). +/// +/// Publishes to every live relay (accepted if any accepts), discovers from all +/// of them de-duplicated by event id, and simply skips relays that are down — +/// so one flaky community relay never breaks sharing. +class RelayPool implements NostrChannel { + RelayPool._(this._connections, this.identity); + + final List _connections; + final NostrIdentity identity; + + @override + String get privateKeyHex => identity.privateKeyHex; + @override + String get publicKeyHex => identity.publicKeyHex; + + /// Number of relays currently connected (for diagnostics/tests). + int get relayCount => _connections.length; + + /// Connects to each of [relayUrls], skipping any that fail. Throws + /// [StateError] only when NONE are reachable, so the caller can treat that as + /// "offline". + static Future connect( + List relayUrls, { + required NostrIdentity identity, + }) async { + final connections = []; + await Future.wait( + relayUrls.map((url) async { + try { + connections.add( + await NostrConnection.connect(url, identity: identity), + ); + } catch (_) { + // Unreachable relay — skip it, keep the rest. + } + }), + ); + if (connections.isEmpty) { + throw StateError('no relays reachable'); + } + return RelayPool._(connections, identity); + } + + @override + Future<({bool accepted, String message})> publish(Event event) async { + final results = await Future.wait( + _connections.map((c) async { + try { + return await c.publish(event); + } catch (_) { + return (accepted: false, message: 'relay error'); + } + }), + ); + final accepted = results.any((r) => r.accepted); + return (accepted: accepted, message: accepted ? '' : 'no relay accepted'); + } + + @override + Stream subscribe(Filter filter) { + final seen = {}; + final subs = >[]; + late StreamController controller; + controller = StreamController( + onListen: () { + for (final c in _connections) { + subs.add( + c.subscribe(filter).listen( + (e) { + if (seen.add(e.id)) controller.add(e); + }, + onError: (_) {}, // one relay erroring shouldn't kill the merge + ), + ); + } + }, + onCancel: () async { + for (final s in subs) { + await s.cancel(); + } + }, + ); + return controller.stream; + } + + @override + Future> reqOnce(Filter filter) async { + final lists = await Future.wait( + _connections.map((c) async { + try { + return await c.reqOnce(filter); + } catch (_) { + return []; + } + }), + ); + final seen = {}; + final out = []; + for (final list in lists) { + for (final e in list) { + if (seen.add(e.id)) out.add(e); + } + } + return out; + } + + @override + Future close() async { + for (final c in _connections) { + await c.close(); + } + } +} diff --git a/packages/commons_core/test/social/relay_pool_test.dart b/packages/commons_core/test/social/relay_pool_test.dart new file mode 100644 index 0000000..52de12f --- /dev/null +++ b/packages/commons_core/test/social/relay_pool_test.dart @@ -0,0 +1,68 @@ +import 'dart:typed_data'; + +import 'package:commons_core/commons_core.dart'; +import 'package:test/test.dart'; + +import '../support/mini_relay.dart'; + +/// Relay-strategy hardening: publish to several relays, discover deduped, and +/// keep working when one is down. +void main() { + Future idFor(int fill) => + NostrKeyDerivation.deriveFromSeed(Uint8List(32)..fillRange(0, 32, fill)); + + Offer offer(String id, NostrIdentity who) => Offer( + id: id, + authorPubkeyHex: who.publicKeyHex, + summary: 'Tomate', + type: OfferType.gift, + approxGeohash: 'sp3e9', + ); + + test('publishes to every relay and discovers deduped across them', () async { + final r1 = await MiniRelay.start(); + final r2 = await MiniRelay.start(); + final alice = await idFor(1); + final bob = await idFor(2); + + final alicePool = await RelayPool.connect([r1.url, r2.url], identity: alice); + await NostrOfferTransport(alicePool).publish(offer('o1', alice)); + expect(r1.storedCount, 1, reason: 'reached relay 1'); + expect(r2.storedCount, 1, reason: 'reached relay 2'); + + final bobPool = await RelayPool.connect([r1.url, r2.url], identity: bob); + final found = await NostrOfferTransport(bobPool) + .discoverUntilEose(const DiscoveryQuery(geohashPrefix: 'sp3e9')); + expect(found, hasLength(1), reason: 'both relays return it → deduped to one'); + + await alicePool.close(); + await bobPool.close(); + await r1.stop(); + await r2.stop(); + }); + + test('tolerates a dead relay (skips it, keeps the live one)', () async { + final r1 = await MiniRelay.start(); + final alice = await idFor(3); + + // Port 1 is unused → connection refused, skipped. + final pool = + await RelayPool.connect([r1.url, 'ws://127.0.0.1:1'], identity: alice); + expect(pool.relayCount, 1); + + final result = await NostrOfferTransport(pool).publish(offer('o2', alice)); + expect(result.accepted, isTrue); + expect(r1.storedCount, 1); + + await pool.close(); + await r1.stop(); + }); + + test('throws when no relay is reachable (caller treats as offline)', () async { + final alice = await idFor(4); + expect( + RelayPool.connect(['ws://127.0.0.1:1'], identity: alice), + throwsA(isA()), + ); + }); +}