fix: event source tracking

packages/ndk/lib/domain_layer/entities/broadcast_state.dart

@@ -51,7 +51,7 @@ class BroadcastState {
 
   /// [networkController] used by relay manger to write responses
   StreamController<RelayBroadcastResponse> networkController =
-      StreamController<RelayBroadcastResponse>();
+      StreamController<RelayBroadcastResponse>.broadcast();
 
   /// stream of state updates \
   /// updates are sent when a relay responds, the whole state is sent \

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);

packages/ndk/lib/domain_layer/usecases/relay_manager.dart

@@ -501,6 +501,22 @@ class RelayManager<T> {
       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<T> {
       }
 
       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

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<Nip01Event>();
-    final seenEventIds = <String>{};
+    final seenEvents = <String, Set<String>>{}; // event_id -> sources
 
     Future<void> paginate() async {
       final since = filter.since;
@@ -373,9 +374,20 @@ class Requests {
       final relayState = <String, _RelayPaginationState>{};
 
       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) {

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<String> _trackingSet;
+  final Map<String, Set<String>> _trackingMap; // event_id -> set of sources
   final List<Stream<Nip01Event>> _inputStreams;
   final StreamController<Nip01Event> _outController;
   final List<EventFilter> _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<String> trackingSet,
     required List<Stream<Nip01Event>> inputStreams,
     required StreamController<Nip01Event> outController,
     required List<EventFilter> 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<Nip01Event> 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<String>.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<String> 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<void> _canClose() async {
     _closedStreams++;

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,

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([

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);
+  });
+}

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));
+  });
+}