From 2a6b4b09474c4f253f308ef91dbc4367c2e701be Mon Sep 17 00:00:00 2001 From: vjrj Date: Fri, 10 Jul 2026 10:35:39 +0200 Subject: [PATCH] =?UTF-8?q?feat(block2):=20relay=20pool=20=E2=80=94=20publ?= =?UTF-8?q?ish=20to=20many=20relays,=20discover=20deduped,=20tolerate=20fa?= =?UTF-8?q?ilures?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Relay-strategy hardening (network-trust §3): don't hang on one relay. - NostrChannel interface: what the transports need (sign, publish, subscribe, reqOnce). Implemented by NostrConnection (single relay) and the new RelayPool. - RelayPool: connects to all configured relays (skips unreachable ones, throws only when none are reachable), publishes to every live relay (accepted if any accepts), merges discovery deduped by event id, and survives a relay dropping. - Transports now depend on NostrChannel, not a concrete connection. - SocialService.openSession takes ALL configured relays (was relays.first); createOffersCubit passes the whole list. commons_core 63 tests green (3 new pool tests over two in-process relays); app_seeds social/market 18 green. --- .../lib/services/social_service.dart | 28 ++-- apps/app_seeds/lib/state/offers_cubit.dart | 4 +- packages/commons_core/lib/commons_core.dart | 2 + .../lib/src/social/nostr/nostr_channel.dart | 21 +++ .../src/social/nostr/nostr_connection.dart | 9 +- .../social/nostr/nostr_message_transport.dart | 4 +- .../social/nostr/nostr_offer_transport.dart | 7 +- .../social/nostr/nostr_trust_transport.dart | 4 +- .../lib/src/social/nostr/relay_pool.dart | 123 ++++++++++++++++++ .../test/social/relay_pool_test.dart | 68 ++++++++++ 10 files changed, 246 insertions(+), 24 deletions(-) create mode 100644 packages/commons_core/lib/src/social/nostr/nostr_channel.dart create mode 100644 packages/commons_core/lib/src/social/nostr/relay_pool.dart create mode 100644 packages/commons_core/test/social/relay_pool_test.dart 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()), + ); + }); +}