diff --git a/packages/ndk/lib/domain_layer/entities/broadcast_state.dart b/packages/ndk/lib/domain_layer/entities/broadcast_state.dart index 75f093dd4..99af89c95 100644 --- a/packages/ndk/lib/domain_layer/entities/broadcast_state.dart +++ b/packages/ndk/lib/domain_layer/entities/broadcast_state.dart @@ -51,7 +51,7 @@ class BroadcastState { /// [networkController] used by relay manger to write responses StreamController networkController = - StreamController(); + StreamController.broadcast(); /// stream of state updates \ /// updates are sent when a relay responds, the whole state is sent \ diff --git a/packages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dart b/packages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dart index 067275520..987497bd2 100644 --- a/packages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dart +++ b/packages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dart @@ -71,6 +71,25 @@ class Broadcast { // register broadcast state _globalState.inFlightBroadcasts[nostrEvent.id] = broadcastState; + // Listen for broadcast responses to update cache sources + broadcastState.networkController.stream.listen((response) async { + if (response.broadcastSuccessful) { + final currentEvent = broadcastState.event; + if (currentEvent != null) { + final cachedEvent = await _cacheManager.loadEvent(currentEvent.id); + if (cachedEvent != null) { + final updatedSources = { + ...cachedEvent.sources, + response.relayUrl + }.toList(); + final updatedEvent = + cachedEvent.copyWith(sources: updatedSources); + await _cacheManager.saveEvent(updatedEvent); + } + } + } + }); + // save event to cache if enabled if (mySaveToCache) { _cacheManager.saveEvent(nostrEvent); diff --git a/packages/ndk/lib/domain_layer/usecases/relay_manager.dart b/packages/ndk/lib/domain_layer/usecases/relay_manager.dart index fb155bb56..89fe754e2 100644 --- a/packages/ndk/lib/domain_layer/usecases/relay_manager.dart +++ b/packages/ndk/lib/domain_layer/usecases/relay_manager.dart @@ -501,6 +501,22 @@ class RelayManager { if (globalState.inFlightBroadcasts[eventId] != null && !globalState .inFlightBroadcasts[eventId]!.networkController.isClosed) { + // Update cache with source if broadcast was successful + if (success) { + final broadcastState = globalState.inFlightBroadcasts[eventId]; + final event = broadcastState?.event; + if (event != null) { + // Merge existing sources with new relay URL, avoiding duplicates + final updatedSources = { + ...event.sources, + relayConnectivity.url + }.toList(); + final updatedEvent = event.copyWith(sources: updatedSources); + // Update the event in broadcast state + broadcastState!.event = updatedEvent; + } + } + globalState.inFlightBroadcasts[eventId]?.networkController.add( RelayBroadcastResponse( relayUrl: relayConnectivity.url, @@ -659,7 +675,7 @@ class RelayManager { } final eventWithSources = - event.copyWith(sources: [...event.sources, connectivity.url]); + event.copyWith(sources: {...event.sources, connectivity.url}.toList()); if (state.networkController.isClosed) { // this might happen because relays even after we send a CLOSE subscription.id, they'll still send more events diff --git a/packages/ndk/lib/domain_layer/usecases/requests/requests.dart b/packages/ndk/lib/domain_layer/usecases/requests/requests.dart index 946eaed06..a5e7c3eba 100644 --- a/packages/ndk/lib/domain_layer/usecases/requests/requests.dart +++ b/packages/ndk/lib/domain_layer/usecases/requests/requests.dart @@ -275,6 +275,7 @@ class Requests { trackingSet: state.returnedIds, outController: state.controller, eventOutFilters: _eventOutFilters, + cacheManager: _cacheWrite.cacheManager, )(); // Record fetched ranges when network requests complete (EOSE received) @@ -345,7 +346,7 @@ class Requests { }) { final requestId = '$name-paginated-${Helpers.getRandomString(10)}'; final aggregatedController = ReplaySubject(); - final seenEventIds = {}; + final seenEvents = >{}; // event_id -> sources Future paginate() async { final since = filter.since; @@ -373,9 +374,20 @@ class Requests { final relayState = {}; for (final event in initialEvents) { - if (!seenEventIds.contains(event.id)) { - seenEventIds.add(event.id); + final existingSources = seenEvents[event.id]; + if (existingSources == null) { + // First time seeing this event + seenEvents[event.id] = event.sources.toSet(); aggregatedController.add(event); + } else { + // Merge sources if this event has new sources + if (event.sources.isNotEmpty) { + final newSources = existingSources..addAll(event.sources); + if (newSources.length > (seenEvents[event.id]?.length ?? 0)) { + seenEvents[event.id] = newSources; + aggregatedController.add(event.copyWith(sources: newSources.toList())); + } + } } // Track oldest timestamp per relay @@ -453,9 +465,20 @@ class Requests { int? oldestTimestamp; for (final event in pageEvents) { - if (!seenEventIds.contains(event.id)) { - seenEventIds.add(event.id); + final existingSources = seenEvents[event.id]; + if (existingSources == null) { + // First time seeing this event + seenEvents[event.id] = event.sources.toSet(); aggregatedController.add(event); + } else { + // Merge sources if this event has new sources + if (event.sources.isNotEmpty) { + final newSources = existingSources..addAll(event.sources); + if (newSources.length > (seenEvents[event.id]?.length ?? 0)) { + seenEvents[event.id] = newSources; + aggregatedController.add(event.copyWith(sources: newSources.toList())); + } + } } // Track oldest timestamp for this relay if (oldestTimestamp == null || event.createdAt < oldestTimestamp) { diff --git a/packages/ndk/lib/domain_layer/usecases/stream_response_cleaner/stream_response_cleaner.dart b/packages/ndk/lib/domain_layer/usecases/stream_response_cleaner/stream_response_cleaner.dart index 7a296f9eb..f0c8ed971 100644 --- a/packages/ndk/lib/domain_layer/usecases/stream_response_cleaner/stream_response_cleaner.dart +++ b/packages/ndk/lib/domain_layer/usecases/stream_response_cleaner/stream_response_cleaner.dart @@ -2,14 +2,16 @@ import 'dart:async'; import '../../../shared/logger/logger.dart'; import '../../entities/event_filter.dart'; import '../../entities/nip_01_event.dart'; +import '../../repositories/cache_manager.dart'; /// given a stream with Nip01 events it tracks the id and adds the one to the provided stream controller \ /// tracking of the happens in the tracking list class StreamResponseCleaner { - final Set _trackingSet; + final Map> _trackingMap; // event_id -> set of sources final List> _inputStreams; final StreamController _outController; final List _eventOutFilters; + final CacheManager? _cacheManager; int get _numStreams => _inputStreams.length; @@ -18,16 +20,19 @@ class StreamResponseCleaner { /// - [trackingSet] a set of ids that are already returned \ /// - [inputStreams] a list of streams that are be listened to \ /// - [outController] the controller that is used to add the events to \ + /// - [cacheManager] optional cache manager to persist event sources \ StreamResponseCleaner({ required Set trackingSet, required List> inputStreams, required StreamController outController, required List eventOutFilters, - }) : _trackingSet = trackingSet, + CacheManager? cacheManager, + }) : _trackingMap = {for (final id in trackingSet) id: {}}, _outController = outController, _inputStreams = inputStreams, - _eventOutFilters = eventOutFilters; + _eventOutFilters = eventOutFilters, + _cacheManager = cacheManager; void call() { for (final stream in _inputStreams) { @@ -37,16 +42,30 @@ class StreamResponseCleaner { void _addStreamListener(Stream stream) { stream.listen((event) { - // check if event id is in the set - if (_trackingSet.contains(event.id)) { + if (_outController.isClosed) { return; } - if (_outController.isClosed) { + // check if event id is already seen + final existingSources = _trackingMap[event.id]; + if (existingSources != null) { + // Event already seen - merge sources if this event has new sources + if (event.sources.isNotEmpty) { + final newSources = Set.from(existingSources)..addAll(event.sources); + // Only emit if we have new sources to add + if (newSources.length > existingSources.length) { + _trackingMap[event.id] = newSources; + final mergedEvent = event.copyWith(sources: newSources.toList()); + _outController.add(mergedEvent); + // Update cache with merged sources + _updateCacheSources(event.id, newSources); + } + } return; } - _trackingSet.add(event.id); + // First time seeing this event + _trackingMap[event.id] = event.sources.toSet(); // check against filters for (final filter in _eventOutFilters) { @@ -63,6 +82,19 @@ class StreamResponseCleaner { }); } + /// Updates the cache with merged sources for an event + void _updateCacheSources(String eventId, Set sources) { + if (_cacheManager == null) return; + + // Load existing event from cache and update sources + _cacheManager.loadEvent(eventId).then((cachedEvent) { + if (cachedEvent != null) { + final updatedEvent = cachedEvent.copyWith(sources: sources.toList()); + _cacheManager.saveEvent(updatedEvent); + } + }); + } + /// used to wait on all streams Future _canClose() async { _closedStreams++; diff --git a/packages/ndk/lib/presentation_layer/init.dart b/packages/ndk/lib/presentation_layer/init.dart index bb743bb9c..b0689b667 100644 --- a/packages/ndk/lib/presentation_layer/init.dart +++ b/packages/ndk/lib/presentation_layer/init.dart @@ -141,7 +141,6 @@ class Initialization { eagerAuth: _ndkConfig.eagerAuth, authCallbackTimeout: _ndkConfig.authCallbackTimeout, ); - engine = JitEngine( cache: _ndkConfig.cache, ignoreRelays: _ndkConfig.ignoreRelays, diff --git a/packages/ndk/test/mocks/mock_relay.dart b/packages/ndk/test/mocks/mock_relay.dart index 5bc37d565..b4700019e 100644 --- a/packages/ndk/test/mocks/mock_relay.dart +++ b/packages/ndk/test/mocks/mock_relay.dart @@ -33,6 +33,7 @@ class MockRelay { bool sendMalformedEvents; String? customWelcomeMessage; int? maxEventsPerRequest; + String? bannedWord; // NIP-46 Remote Signer Support static const int kNip46Kind = BunkerRequest.kKind; @@ -58,6 +59,7 @@ class MockRelay { this.sendMalformedEvents = false, this.customWelcomeMessage, this.maxEventsPerRequest, + this.bannedWord, int? explicitPort, }) : _nip65s = nip65s { if (explicitPort != null) { @@ -147,6 +149,16 @@ class MockRelay { if (eventJson[0] == "EVENT") { Nip01Event newEvent = Nip01EventModel.fromJson(eventJson[1]); if (verify(newEvent.pubKey, newEvent.id, newEvent.sig!)) { + // Check if event contains banned word + if (bannedWord != null && newEvent.content.contains(bannedWord!)) { + webSocket.add(jsonEncode([ + "OK", + newEvent.id, + false, + "blocked: content contains banned word" + ])); + return; + } // Check auth for events if required (any authenticated user is OK) if (requireAuthForEvents && authenticatedPubkeys.isEmpty) { webSocket.add(jsonEncode([ diff --git a/packages/ndk/test/usecases/broadcast_sources_test.dart b/packages/ndk/test/usecases/broadcast_sources_test.dart new file mode 100644 index 000000000..edeb18876 --- /dev/null +++ b/packages/ndk/test/usecases/broadcast_sources_test.dart @@ -0,0 +1,43 @@ +import 'package:ndk/ndk.dart'; +import 'package:ndk/shared/nips/nip01/bip340.dart'; +import 'package:test/test.dart'; + +import '../mocks/mock_event_verifier.dart'; +import '../mocks/mock_relay.dart'; + +void main() async { + test("broadcast should update source", () async { + final relay = MockRelay(name: "relay"); + + await relay.startServer(); + addTearDown(() => relay.stopServer()); + + final ndk = Ndk(NdkConfig( + eventVerifier: MockEventVerifier(), + cache: MemCacheManager(), + bootstrapRelays: [relay.url], + )); + addTearDown(() => ndk.destroy()); + + final keypair = Bip340.generatePrivateKey(); + final signer = Bip340EventSigner( + privateKey: keypair.privateKey, + publicKey: keypair.publicKey, + ); + ndk.accounts.loginExternalSigner(signer: signer); + + final event = Nip01Event( + pubKey: keypair.publicKey, + kind: 1, + tags: [], + content: "content", + ); + + await ndk.broadcast.broadcast(nostrEvent: event).broadcastDoneFuture; + + final localEvent = await ndk.config.cache.loadEvent(event.id); + + expect(localEvent, isNotNull); + expect(localEvent!.sources, isNotEmpty); + }); +} diff --git a/packages/ndk/test/usecases/stream_response_cleaner/event_sources_merge_test.dart b/packages/ndk/test/usecases/stream_response_cleaner/event_sources_merge_test.dart new file mode 100644 index 000000000..9a4b42362 --- /dev/null +++ b/packages/ndk/test/usecases/stream_response_cleaner/event_sources_merge_test.dart @@ -0,0 +1,62 @@ +import 'package:ndk/shared/nips/nip01/bip340.dart'; +import 'package:test/test.dart'; +import 'package:ndk/ndk.dart'; + +import '../../mocks/mock_event_verifier.dart'; +import '../../mocks/mock_relay.dart'; + +void main() async { + test("requests should update sources", () async { + final bannedWord = "cow"; + + final relay1 = MockRelay(name: "relay 1"); + final relay2 = MockRelay(name: "relay 2"); + final relay3 = MockRelay(name: "relay 3", bannedWord: bannedWord); + + await relay1.startServer(); + await relay2.startServer(); + await relay3.startServer(); + + addTearDown(() async { + await relay1.stopServer(); + await relay2.stopServer(); + await relay3.stopServer(); + }); + + final ndk = Ndk(NdkConfig( + eventVerifier: MockEventVerifier(), + cache: MemCacheManager(), + bootstrapRelays: [relay1.url, relay2.url, relay3.url], + )); + + addTearDown(() => ndk.destroy()); + + final keypair = Bip340.generatePrivateKey(); + final signer = Bip340EventSigner( + privateKey: keypair.privateKey, + publicKey: keypair.publicKey, + ); + ndk.accounts.loginExternalSigner(signer: signer); + + final event = Nip01Event( + pubKey: keypair.publicKey, + kind: 1, + tags: [], + content: bannedWord, + ); + await ndk.broadcast.broadcast(nostrEvent: event).broadcastDoneFuture; + + await ndk.config.cache.clearAll(); + + final query = ndk.requests.query(filter: Filter(ids: [event.id])); + final events = await query.future; + + // The last event should have all merged sources + expect(events.last.sources.length, equals(2)); + + final localEvent = await ndk.config.cache.loadEvent(event.id); + + expect(localEvent, isNotNull); + expect(localEvent!.sources.length, equals(2)); + }); +}