Add pull to refresh feature in DMs

Closes: https://github.com/damus-io/damus/issues/3352
Changelog-Added: Added a pull to refresh feature on DMs that allows users to resync DMs with their relays
Signed-off-by: Daniel D’Aquino <daniel@daquino.me>

damus/ContentView.swift

@@ -180,7 +180,7 @@ struct ContentView: View {
                 NotificationsView(state: damus, notifications: home.notifications, subtitle: $menu_subtitle)
                 
             case .dms:
-                DirectMessagesView(damus_state: damus_state!, model: damus_state!.dms, settings: damus_state!.settings, subtitle: $menu_subtitle)
+                DirectMessagesView(damus_state: damus_state!, home: home, model: damus_state!.dms, settings: damus_state!.settings, subtitle: $menu_subtitle)
             }
         }
         .background(DamusColors.adaptableWhite)

damus/Core/Nostr/RelayConnection.swift

@@ -270,7 +270,7 @@ final class RelayConnection: ObservableObject {
                 }
                 return
             }
-            print("failed to decode event \(messageString)")
+            print("\(self.relay_url): failed to decode event \(messageString)")
         case .data(let messageData):
             if let messageString = String(data: messageData, encoding: .utf8) {
                 await receive(message: .string(messageString))
@@ -299,7 +299,7 @@ final class RelayConnection: ObservableObject {
                 throw NegentropySyncError.notSupported
             }
         }
-        let timeout = timeout ?? .seconds(5)
+        let timeout = timeout ?? .seconds(3)
         let frameSizeLimit = 60_000 // Copied from rust-nostr project: Default frame limit is 128k. Halve that (hex encoding) and subtract a bit (JSON msg overhead)
         try? negentropyVector.seal()    // Error handling note: We do not care if it throws an `alreadySealed` error. As long as it is sealed in the end it is fine
         let negentropyClient = try Negentropy(storage: negentropyVector, frameSizeLimit: frameSizeLimit)

damus/Core/Nostr/RelayPool.swift

@@ -661,10 +661,12 @@ class RelayPool {
                     }
                 }
                 catch {
-                    if let negentropyError = error as? RelayConnection.NegentropySyncError,
-                       case .notSupported = negentropyError,
-                       ignoreUnsupportedRelays {
+                    if ignoreUnsupportedRelays {
                         // Do not throw error, ignore the relays that do not support negentropy
+                        // Note: Some relays such as wss://nos.lol/v2 advertise negentropy but throw an error such as `["NOTICE","ERROR: bad msg: negentropy disabled"]`
+                        // Therefore, realistically, we cannot rely on what the relay advertises and
+                        // we have to suppress those errors if we want to ignore unsupported relays to avoid the whole multi-relay negentropy syncing operation to fail
+                        Log.error("Error while negentropy streaming: %s", for: .networking, error.localizedDescription)
                     }
                     else {
                         throw error

damus/Features/DMs/Views/DirectMessagesView.swift

@@ -15,6 +15,7 @@ enum DMType: Hashable {
 
 struct DirectMessagesView: View {
     let damus_state: DamusState
+    let home: HomeModel
 
     @State var dm_type: DMType = .friend
     @ObservedObject var model: DirectMessagesModel
@@ -37,6 +38,12 @@ struct DirectMessagesView: View {
             }
             .padding(.horizontal)
         }
+        .refreshable {
+            // Fetch full DM history without the `since` optimization.
+            // This allows users to manually sync older DMs that may have
+            // been missed due to the optimized network filter.
+            await home.fetchFullDMHistory()
+        }
         .padding(.bottom, tabHeight)
     }
     
@@ -136,6 +143,6 @@ func would_filter_non_friends_from_dms(contacts: Contacts, dms: [DirectMessageMo
 struct DirectMessagesView_Previews: PreviewProvider {
     static var previews: some View {
         let ds = test_damus_state
-        DirectMessagesView(damus_state: ds, model: ds.dms, settings: ds.settings, subtitle: .constant(nil))
+        DirectMessagesView(damus_state: ds, home: HomeModel(), model: ds.dms, settings: ds.settings, subtitle: .constant(nil))
     }
 }

damus/Features/Timeline/Models/HomeModel.swift

@@ -898,6 +898,33 @@ class HomeModel: ContactsDelegate, ObservableObject {
         }
     }
 
+    /// Fetches DM history from relays.
+    ///
+    /// By default, the DM subscription uses `optimizeNetworkFilter: true` which adds a
+    /// `since` parameter based on the latest local timestamp. This is efficient but can
+    /// miss older DMs if the local database doesn't have complete history.
+    ///
+    /// This method requests full DM history with negentropy.
+    func fetchFullDMHistory() async {
+        // DMs sent to us (limit to prevent runaway pulls; user can pull again for more)
+        var dms_filter = NostrFilter(kinds: [.dm])
+        dms_filter.pubkeys = [damus_state.pubkey]
+        dms_filter.limit = 500
+
+        // DMs we sent
+        var our_dms_filter = NostrFilter(kinds: [.dm])
+        our_dms_filter.authors = [damus_state.pubkey]
+        our_dms_filter.limit = 500
+
+        let filters = [dms_filter, our_dms_filter]
+        let timeoutSeconds: UInt64 = 20
+        
+        for await lender in self.damus_state.nostrNetwork.reader.streamExistingEvents(filters: filters, timeout: .seconds(timeoutSeconds), streamMode: .ndbAndNetworkParallel(networkOptimization: .negentropy)) {
+            if Task.isCancelled { return }
+            lender.justUseACopy({ self.process_event(ev: $0, context: .other) })
+        }
+    }
+
     @MainActor
     func handle_dm(_ ev: NostrEvent) {
         guard should_show_event(state: damus_state, ev: ev) else {

damusTests/SubscriptionManagerNegentropyTests.swift

@@ -134,7 +134,7 @@ final class SubscriptionManagerNegentropyTests: XCTestCase {
         negentropyEventExpectations: [NoteId: XCTestExpectation],
         ndbEoseExpectation: XCTestExpectation? = nil,
         networkEoseExpectation: XCTestExpectation? = nil,
-        eoseExpectation: XCTestExpectation? = nil
+        eoseExpectation: XCTestExpectation? = nil,
     ) {
         Task {
             var ndbEoseSeen = false
@@ -467,6 +467,51 @@ final class SubscriptionManagerNegentropyTests: XCTestCase {
         // (Order not enforced because we don't make guarantees on the order of A/C and B/D
         await fulfillment(of: [getsNoteAFromNdb, getsNoteCFromNdb, ndbEose, getsNoteBFromNegentropy, getsNoteDFromNegentropy, networkEose], timeout: 10.0)
     }
+    
+    func testPartialUnsupportedRelayPool() async throws {
+        // Given: Two relays (one with negentropy, another one not), and the one with negentropy has an event we need
+        let relay2 = try await setupRelay(port: 9092)
+        
+        let relayUrl1 = RelayURL("ws://nos.lol/v2")!    // This can be any relay that does not support negentropy
+                                                        // Adding an external relay may cause flakiness if the relay enables negentropy, but currently
+                                                        // there is no feasible way to configure a local relay that rejects negentropy requests.
+                                                        // Therefore, keep this external relay until it causes issues and then we can investigate
+                                                        // how to improve this test's robustness.
+        let relayUrl2 = RelayURL(await relay2.url().description)!
+        
+        let noteA = NostrEvent(content: "A", keypair: test_keypair)!
+        let noteB = NostrEvent(content: "B", keypair: test_keypair)!
+        
+        // Connect to relay1 and send noteA + noteB
+        let relayConnection2 = await connectToRelay(url: relayUrl2, label: "Relay1")
+        sendEvents([noteA, noteB], to: relayConnection2)
+        
+        let ndb = await test_damus_state.ndb
+        storeEventsInNdb([noteB], ndb: ndb)
+        
+        let networkManager = try await setupNetworkManager(with: [relayUrl1, relayUrl2], ndb: ndb)
+        
+        let getsNoteBFromNdb = XCTestExpectation(description: "Gets note B from NDB before ndbEose")
+        let getsNoteAFromNegentropy = XCTestExpectation(description: "Gets note A via negentropy after ndbEose")
+        let ndbEose = XCTestExpectation(description: "Receives NDB EOSE")
+        let networkEose = XCTestExpectation(description: "Receives network EOSE")
+        let generalEose = XCTestExpectation(description: "Receives general EOSE")
+        
+        // When: Using negentropy streaming mode across two relays
+        runAdvancedStream(
+            networkManager: networkManager,
+            filters: [NostrFilter(kinds: [.text])],
+            streamMode: .ndbAndNetworkParallel(networkOptimization: .negentropy),
+            ndbEventExpectations: [noteB.id: getsNoteBFromNdb],
+            negentropyEventExpectations: [noteA.id: getsNoteAFromNegentropy],
+            ndbEoseExpectation: ndbEose,
+            networkEoseExpectation: networkEose,
+            eoseExpectation: generalEose
+        )
+        
+        // Then: Should receive noteB from NDB, then ndbEose, then noteA via negentropy
+        await fulfillment(of: [getsNoteBFromNdb, ndbEose, getsNoteAFromNegentropy, networkEose, generalEose], timeout: 10.0)
+    }
 }
 
 // MARK: - Test Doubles