Merge branch 'main' into nip56-mapping

planetary-social · Jul 16, 2024 · cbcc7c8 · cbcc7c8
2 parents df298ab + 7a0d006
commit cbcc7c8
Show file tree

Hide file tree

Showing 8 changed files with 25 additions and 22 deletions.
diff --git a/Nos/Service/CurrentUser.swift b/Nos/Service/CurrentUser.swift
@@ -193,8 +193,7 @@ import Dependencies
             authorKeys: [key],
             kinds: [.contactList],
             limit: 2, // small hack to make sure this filter doesn't get closed for being stale
-            since: author.lastUpdatedContactList,
-            keepSubscriptionOpen: true
+            since: author.lastUpdatedContactList
         )
         subscriptions.append(
             await relayService.fetchEvents(matching: contactFilter)

diff --git a/Nos/Service/PushNotificationService.swift b/Nos/Service/PushNotificationService.swift
@@ -96,7 +96,7 @@ import Combine
         }
 
         let userMentionsFilter = Filter(
-            kinds: [.text, .longFormContent, .like], 
+            kinds: [.text], 
             pTags: [authorKey], 
             limit: 50,
             keepSubscriptionOpen: true

diff --git a/Nos/Service/Relay/PagedRelaySubscription.swift b/Nos/Service/Relay/PagedRelaySubscription.swift
@@ -44,9 +44,11 @@ class PagedRelaySubscription {
             // progressively older events as we page down.
             var pagedEventsFilter = filter
             pagedEventsFilter.until = startDate
+            pagedEventsFilter.keepSubscriptionOpen = false
             var newEventsFilter = filter
 
             newEventsFilter.since = startDate
+            newEventsFilter.keepSubscriptionOpen = true
             newEventsFilter.limit = nil
             for relayAddress in relayAddresses {
                 newEventsSubscriptionIDs.insert(

diff --git a/Nos/Service/Relay/RelayService.swift b/Nos/Service/Relay/RelayService.swift
@@ -108,9 +108,10 @@ extension RelayService {
         }
     }
 
-    private func sendClose(from client: WebSocketClient, subscription: String) {
+    private func sendClose(from client: WebSocketClient, subscriptionID: RelaySubscription.ID) async {
         do {
-            let request: [Any] = ["CLOSE", subscription]
+            await subscriptionManager.forceCloseSubscriptionCount(for: subscriptionID)
+            let request: [Any] = ["CLOSE", subscriptionID]
             let requestData = try JSONSerialization.data(withJSONObject: request)
             let requestString = String(data: requestData, encoding: .utf8)!
             client.write(string: requestString)
@@ -120,15 +121,18 @@ extension RelayService {
     }
 
     private func sendCloseToAll(for subscription: RelaySubscription.ID) async {
-        await subscriptionManager.sockets().forEach { self.sendClose(from: $0, subscription: subscription) }
+        let sockets = await subscriptionManager.sockets()
+        for socket in sockets {
+            await self.sendClose(from: socket, subscriptionID: subscription) 
+        }
         Task { await processSubscriptionQueue() }
     }
 
     func closeConnection(to relayAddress: String?) async {
         guard let address = relayAddress else { return }
         if let socket = await subscriptionManager.socket(for: address) {
             for subscription in await subscriptionManager.active() {
-                self.sendClose(from: socket, subscription: subscription.id)
+                await self.sendClose(from: socket, subscriptionID: subscription.id)
             }
 
             await subscriptionManager.close(socket: socket)
@@ -330,7 +334,7 @@ extension RelayService {
             subscription.closesAfterResponse {
             Log.debug("\(socket.host) has finished responding on \(subID). Closing subscription.")
             // This is a one-off request. Close it.
-            sendClose(from: socket, subscription: subID)
+            await sendClose(from: socket, subscriptionID: subID)
         }
     }
 
@@ -355,7 +359,7 @@ extension RelayService {
             let jsonEvent = try JSONDecoder().decode(JSONEvent.self, from: jsonData)
             await self.parseQueue.push(jsonEvent, from: socket)
 
-            if var subscription = await subscriptionManager.subscription(from: subscriptionID) {
+            if let subscription = await subscriptionManager.subscription(from: subscriptionID) {
                 subscription.receivedEventCount += 1
                 subscription.events.send(jsonEvent)
                 if subscription.closesAfterResponse {
@@ -805,21 +809,21 @@ extension RelayService: WebSocketDelegate {
 
         Task {
             switch event {
-            case .connected:
+            case .connected, .viabilityChanged(true):
                 await handleConnection(from: client)
-            case .disconnected(let reason, let code):
+            case .disconnected:
+                await subscriptionManager.remove(socket)
+            case .peerClosed:
                 await subscriptionManager.remove(socket)
-                print("websocket is disconnected: \(reason) with code: \(code)")
             case .text(let string):
                 await parseResponse(string, socket)
             case .binary:
                 break
-            case .ping, .pong, .viabilityChanged, .reconnectSuggested, .peerClosed:
+            case .ping, .pong, .viabilityChanged, .reconnectSuggested:
                 break
             case .cancelled:
                 await subscriptionManager.trackError(socket: socket)
                 await subscriptionManager.remove(socket)
-                print("websocket is cancelled")
             case .error(let error):
                 await subscriptionManager.trackError(socket: socket)
                 await subscriptionManager.remove(socket)

diff --git a/Nos/Service/Relay/RelaySubscriptionManager.swift b/Nos/Service/Relay/RelaySubscriptionManager.swift
@@ -144,6 +144,9 @@ actor RelaySubscriptionManagerActor: RelaySubscriptionManager {
         if let index = sockets.firstIndex(where: { $0 === socket }) {
             sockets.remove(at: index)
         }
+        all.removeAll { subscription in
+            subscription.relayAddress == socket.url
+        }
     }
 
     func socket(for address: String) -> WebSocket? {

diff --git a/Nos/Views/Home/HomeFeedView.swift b/Nos/Views/Home/HomeFeedView.swift
@@ -84,8 +84,7 @@ struct HomeFeedView: View {
             let textFilter = Filter(
                 authorKeys: followedKeys, 
                 kinds: [.text, .delete, .repost, .longFormContent, .report], 
-                since: storiesCutoffDate,
-                keepSubscriptionOpen: true
+                since: storiesCutoffDate
             )
             let textSubs = await relayService.fetchEvents(matching: textFilter)
             relaySubscriptions.append(textSubs)

diff --git a/Nos/Views/NotificationsView.swift b/Nos/Views/NotificationsView.swift
@@ -41,7 +41,7 @@ struct NotificationsView: View {
             kinds: [.text], 
             pTags: [currentUserKey], 
             limit: 100,
-            keepSubscriptionOpen: true
+            keepSubscriptionOpen: false
         )
         let subscriptions = await relayService.fetchEvents(matching: filter)
         relaySubscriptions.append(subscriptions)

diff --git a/Nos/Views/Profile/ProfileFeedType.swift b/Nos/Views/Profile/ProfileFeedType.swift
@@ -18,10 +18,6 @@ enum ProfileFeedType {
             kinds = [.text, .delete]
         }
 
-        return Filter(
-            authorKeys: [author.hexadecimalPublicKey ?? "error"],
-            kinds: kinds,
-            keepSubscriptionOpen: true
-        )
+        return Filter(authorKeys: [author.hexadecimalPublicKey ?? "error"], kinds: kinds)
     }
 }