Add ndb subscription tests

Signed-off-by: Daniel D’Aquino <daniel@daquino.me>

damus.xcodeproj/project.pbxproj

@@ -1754,6 +1754,8 @@
 		D7DF58342DFCF18D00E9AD28 /* SendPaymentView.swift in Sources */ = {isa = PBXBuildFile; fileRef = D7DF58312DFCF18800E9AD28 /* SendPaymentView.swift */; };
 		D7EB00B02CD59C8D00660C07 /* PresentFullScreenItemNotify.swift in Sources */ = {isa = PBXBuildFile; fileRef = D7EB00AF2CD59C8300660C07 /* PresentFullScreenItemNotify.swift */; };
 		D7EB00B12CD59C8D00660C07 /* PresentFullScreenItemNotify.swift in Sources */ = {isa = PBXBuildFile; fileRef = D7EB00AF2CD59C8300660C07 /* PresentFullScreenItemNotify.swift */; };
+		D7EBF8BB2E59022A004EAE29 /* NostrNetworkManagerTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = D7EBF8BA2E5901F7004EAE29 /* NostrNetworkManagerTests.swift */; };
+		D7EBF8BE2E59470D004EAE29 /* test_notes.jsonl in Resources */ = {isa = PBXBuildFile; fileRef = D7EBF8BD2E594708004EAE29 /* test_notes.jsonl */; };
 		D7EDED152B11776B0018B19C /* LibreTranslateServer.swift in Sources */ = {isa = PBXBuildFile; fileRef = 3AE45AF5297BB2E700C1D842 /* LibreTranslateServer.swift */; };
 		D7EDED162B1177840018B19C /* LNUrls.swift in Sources */ = {isa = PBXBuildFile; fileRef = 4CB883B5297730E400DC99E7 /* LNUrls.swift */; };
 		D7EDED172B1177960018B19C /* TranslationService.swift in Sources */ = {isa = PBXBuildFile; fileRef = 3AAA95C9298DF87B00F3D526 /* TranslationService.swift */; };
@@ -2688,6 +2690,8 @@
 		D7DEEF2E2A8C021E00E0C99F /* NostrEventTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NostrEventTests.swift; sourceTree = "<group>"; };
 		D7DF58312DFCF18800E9AD28 /* SendPaymentView.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = SendPaymentView.swift; sourceTree = "<group>"; };
 		D7EB00AF2CD59C8300660C07 /* PresentFullScreenItemNotify.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = PresentFullScreenItemNotify.swift; sourceTree = "<group>"; };
+		D7EBF8BA2E5901F7004EAE29 /* NostrNetworkManagerTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NostrNetworkManagerTests.swift; sourceTree = "<group>"; };
+		D7EBF8BD2E594708004EAE29 /* test_notes.jsonl */ = {isa = PBXFileReference; lastKnownFileType = text; path = test_notes.jsonl; sourceTree = "<group>"; };
 		D7EDED1B2B1178FE0018B19C /* NoteContent.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NoteContent.swift; sourceTree = "<group>"; };
 		D7EDED1D2B11797D0018B19C /* LongformEvent.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = LongformEvent.swift; sourceTree = "<group>"; };
 		D7EDED202B117DCA0018B19C /* SequenceUtils.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = SequenceUtils.swift; sourceTree = "<group>"; };
@@ -3670,6 +3674,7 @@
 		4CE6DEF627F7A08200C66700 /* damusTests */ = {
 			isa = PBXGroup;
 			children = (
+				D7EBF8BC2E5946F9004EAE29 /* NostrNetworkManagerTests */,
 				D7DB1FED2D5AC50F00CF06DA /* NIP44v2EncryptionTests.swift */,
 				D7A0D8742D1FE66A00DCBE59 /* EditPictureControlTests.swift */,
 				E06336A72B7582D600A88E6B /* Assets */,
@@ -4991,6 +4996,15 @@
 			path = NIP65;
 			sourceTree = "<group>";
 		};
+		D7EBF8BC2E5946F9004EAE29 /* NostrNetworkManagerTests */ = {
+			isa = PBXGroup;
+			children = (
+				D7EBF8BD2E594708004EAE29 /* test_notes.jsonl */,
+				D7EBF8BA2E5901F7004EAE29 /* NostrNetworkManagerTests.swift */,
+			);
+			path = NostrNetworkManagerTests;
+			sourceTree = "<group>";
+		};
 		E06336A72B7582D600A88E6B /* Assets */ = {
 			isa = PBXGroup;
 			children = (
@@ -5313,6 +5327,7 @@
 			files = (
 				E06336AB2B75850100A88E6B /* img_with_location.jpeg in Resources */,
 				D7DB1FF12D5AC5D700CF06DA /* nip44.vectors.json in Resources */,
+				D7EBF8BE2E59470D004EAE29 /* test_notes.jsonl in Resources */,
 				4C0C039A2A61E27B0098B3B8 /* bool_setting.wasm in Resources */,
 				D7DB1FF32D5AC5EA00CF06DA /* LICENSES in Resources */,
 				4C0C03992A61E27B0098B3B8 /* primal.wasm in Resources */,
@@ -5924,6 +5939,7 @@
 				D71DC1EC2A9129C3006E207C /* PostViewTests.swift in Sources */,
 				3AAC7A022A60FE72002B50DF /* LocalizationUtilTests.swift in Sources */,
 				D7CBD1D62B8D509800BFD889 /* DamusPurpleImpendingExpirationTests.swift in Sources */,
+				D7EBF8BB2E59022A004EAE29 /* NostrNetworkManagerTests.swift in Sources */,
 				D7DEEF2F2A8C021E00E0C99F /* NostrEventTests.swift in Sources */,
 				4C8D00D429E3C5D40036AF10 /* NIP19Tests.swift in Sources */,
 				3A30410129AB12AA008A0F29 /* EventGroupViewTests.swift in Sources */,

damus/Core/Networking/NostrNetworkManager/SubscriptionManager.swift

@@ -33,23 +33,28 @@ extension NostrNetworkManager {
         func subscribe(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil) -> AsyncStream<StreamItem> {
             return AsyncStream<StreamItem> { continuation in
                 let ndbStreamTask = Task {
-                    for await item in try self.ndb.subscribe(filters: try filters.map({ try NdbFilter(from: $0) })) {
+                    do {
-                        switch item {
+                        for await item in try self.ndb.subscribe(filters: try filters.map({ try NdbFilter(from: $0) })) {
-                        case .eose:
+                            switch item {
-                            continuation.yield(.eose)
+                            case .eose:
-                        case .event(let noteKey):
+                                continuation.yield(.eose)
-                            let lender: NdbNoteLender = { lend in
+                            case .event(let noteKey):
-                                guard let ndbNoteTxn = self.ndb.lookup_note_by_key(noteKey) else {
+                                let lender: NdbNoteLender = { lend in
-                                    throw NdbNoteLenderError.errorLoadingNote
+                                    guard let ndbNoteTxn = self.ndb.lookup_note_by_key(noteKey) else {
+                                        throw NdbNoteLenderError.errorLoadingNote
+                                    }
+                                    guard let unownedNote = UnownedNdbNote(ndbNoteTxn) else {
+                                        throw NdbNoteLenderError.errorLoadingNote
+                                    }
+                                    lend(unownedNote)
                                 }
-                                guard let unownedNote = UnownedNdbNote(ndbNoteTxn) else {
+                                continuation.yield(.event(borrow: lender))
-                                    throw NdbNoteLenderError.errorLoadingNote
-                                }
-                                lend(unownedNote)
                             }
-                            continuation.yield(.event(borrow: lender))
                         }
                     }
+                    catch {
+                        Log.error("NDB streaming error: %s", for: .ndb, error.localizedDescription)
+                    }
                 }
                 let streamTask = Task {
                     for await _ in self.pool.subscribe(filters: filters, to: desiredRelays) {

damus/Features/Timeline/Models/HomeModel.swift

@@ -561,8 +561,8 @@ class HomeModel: ContactsDelegate {
                     try? borrow { ev in
                         event = ev.toOwned()
                     }
-                    guard let event else { return }
+                    guard let theEvent = event else { return }
-                    await self.process_event(ev: event, context: .notifications)
+                    await self.process_event(ev: theEvent, context: .notifications)
                 case .eose:
                     guard let txn = NdbTxn(ndb: damus_state.ndb) else { return }
                     load_profiles(context: "notifications", load: .from_keys(notifications.uniq_pubkeys()), damus_state: damus_state, txn: txn)

damusTests/Mocking/MockDamusState.swift

@@ -11,7 +11,8 @@ import EmojiPicker
 
 // Generates a test damus state with configurable mock parameters
 func generate_test_damus_state(
-    mock_profile_info: [Pubkey: Profile]?
+    mock_profile_info: [Pubkey: Profile]?,
+    home: HomeModel? = nil
 ) -> DamusState {
     // Create a unique temporary directory
     let ndb = Ndb.test
@@ -32,7 +33,7 @@ func generate_test_damus_state(
                            boosts: .init(our_pubkey: our_pubkey),
                            contacts: .init(our_pubkey: our_pubkey), mutelist_manager: mutelist_manager,
                            profiles: profiles,
-                           dms: .init(our_pubkey: our_pubkey),
+                           dms: home?.dms ?? .init(our_pubkey: our_pubkey),
                            previews: .init(),
                            zaps: .init(our_pubkey: our_pubkey),
                            lnurls: .init(),
@@ -53,5 +54,7 @@ func generate_test_damus_state(
                            favicon_cache: .init()
     )
     
+    home?.damus_state = damus
+
     return damus
 }

damusTests/NostrNetworkManagerTests/NostrNetworkManagerTests.swift

@@ -0,0 +1,86 @@
+//
+//  NostrNetworkManagerTests.swift
+//  damus
+//
+//  Created by Daniel D'Aquino on 2025-08-22.
+//
+
+import XCTest
+@testable import damus
+
+
+class NostrNetworkManagerTests: XCTestCase {
+    var damusState: DamusState? = nil
+    
+    override func setUpWithError() throws {
+        // Put setup code here. This method is called before the invocation of each test method in the class.
+        damusState = generate_test_damus_state(mock_profile_info: nil)
+
+        let notesJSONL = getTestNotesJSONL()
+
+        for noteText in notesJSONL.split(separator: "\n") {
+            let _ = damusState!.ndb.process_event("[\"EVENT\",\"subid\",\(String(noteText))]")
+        }
+    }
+    
+    override func tearDownWithError() throws {
+        // Put teardown code here. This method is called after the invocation of each test method in the class.
+        damusState = nil
+    }
+    
+    func getTestNotesJSONL() -> String {
+        // Get the path for the test_notes.jsonl file in the same folder as this test file
+        let testBundle = Bundle(for: type(of: self))
+        let fileURL = testBundle.url(forResource: "test_notes", withExtension: "jsonl")!
+
+        // Load the contents of the file
+        return try! String(contentsOf: fileURL, encoding: .utf8)
+    }
+    
+    func ensureSubscribeGetsAllExpectedNotes(filter: NostrFilter, expectedCount: Int) {
+        let endOfStream = XCTestExpectation(description: "Stream should receive EOSE")
+        let gotAtLeastExpectedCount = XCTestExpectation(description: "Stream should receive at least the expected number of items")
+        var receivedCount = 0
+        var eventIds: Set<NoteId> = []
+        Task {
+            for await item in self.damusState!.nostrNetwork.reader.subscribe(filters: [filter]) {
+                switch item {
+                case .event(borrow: let borrow):
+                    try? borrow { event in
+                        receivedCount += 1
+                        if eventIds.contains(event.id) {
+                            XCTFail("Got duplicate event ID: \(event.id) ")
+                        }
+                        eventIds.insert(event.id)
+                    }
+                    if receivedCount == expectedCount {
+                        gotAtLeastExpectedCount.fulfill()
+                    }
+                case .eose:
+                    // End of stream, break out of the loop
+                    endOfStream.fulfill()
+                }
+            }
+        }
+        wait(for: [endOfStream, gotAtLeastExpectedCount], timeout: 10.0)
+        XCTAssertEqual(receivedCount, expectedCount, "Event IDs: \(eventIds.map({ $0.hex() }))")
+    }
+    
+    /// Tests to ensure that subscribing gets the correct amount of events
+    ///
+    /// ## Implementation notes:
+    ///
+    /// To create a new scenario, `nak` can be used as a reference:
+    /// 1. `cd` into the folder where the `test_notes.jsonl` file is
+    /// 2. Run `nak serve --events test_notes.jsonl`
+    /// 3. On a separate terminal, run `nak` commands with the desired filter against the local relay, and get the line count. Example:
+    /// ```
+    /// nak req --kind 1 ws://localhost:10547 | wc -l
+    /// ```
+    func testNdbSubscription() {
+        ensureSubscribeGetsAllExpectedNotes(filter: NostrFilter(kinds: [.text]), expectedCount: 57)
+        ensureSubscribeGetsAllExpectedNotes(filter: NostrFilter(authors: [Pubkey(hex: "32e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245")!]), expectedCount: 22)
+        ensureSubscribeGetsAllExpectedNotes(filter: NostrFilter(kinds: [.boost], referenced_ids: [NoteId(hex: "64b26d0a587f5f894470e1e4783756b4d8ba971226de975ee30ac1b69970d5a1")!]), expectedCount: 5)
+        ensureSubscribeGetsAllExpectedNotes(filter: NostrFilter(kinds: [.text, .boost, .zap], referenced_ids: [NoteId(hex: "64b26d0a587f5f894470e1e4783756b4d8ba971226de975ee30ac1b69970d5a1")!], limit: 500), expectedCount: 5)
+    }
+}