Page past excluded authored-thread replies in push resolution

Signed-off-by: Tom Brow <tomb@block.xyz>

crates/buzz-relay/src/api/bridge.rs

@@ -1467,6 +1467,7 @@ async fn query_events_authed(
             let allowed = [
                 "kinds",
                 "authored_thread_replies",
+                "before_id",
                 "since",
                 "until",
                 "limit",
@@ -8007,6 +8008,22 @@ pub(crate) mod postgres_tests {
             events.iter().map(|e| e.id).collect::<Vec<_>>(),
             vec![reply.id]
         );
+        let cursor_body = serde_json::to_vec(&serde_json::json!([{
+            "kinds": [9], "authored_thread_replies": true, "limit": 1,
+            "until": reply.created_at.as_secs(), "before_id": reply.id.to_hex()
+        }]))
+        .unwrap();
+        let (cursor_status, cursor_bytes) = post_query(
+            state.clone(),
+            &host,
+            &owner.public_key().to_hex(),
+            &cursor_body,
+        )
+        .await;
+        assert_eq!(cursor_status, StatusCode::OK);
+        assert!(serde_json::from_slice::<Vec<nostr::Event>>(&cursor_bytes)
+            .unwrap()
+            .is_empty());
         let (status, bytes) =
             post_query(state.clone(), &host, &other.public_key().to_hex(), body).await;
         assert_eq!(status, StatusCode::OK);

mobile/ios/BuzzPushKit/Sources/BuzzPushKit/BuzzPushNotificationResolver.swift

@@ -142,9 +142,12 @@ public final class BuzzPushNotificationResolver: BuzzPushNotificationResolving {
     _ community: PushLeaseCommunity,
     completion: @escaping ((VerifiedNostrEvent, PushLeaseCommunity)?) -> Void
   ) {
-    let groups = [false, true].map { authored in
-      community.policies.filter { ($0.filter.authoredThreadReplies == true) == authored }
-    }.filter { !$0.isEmpty }
+    let ordinary = community.policies.filter { $0.filter.authoredThreadReplies != true }
+    // Each relational policy needs its own cursor: a union's oldest row could
+    // skip rows hidden by another policy's per-filter limit.
+    let groups = ([ordinary] + community.policies.filter {
+      $0.filter.authoredThreadReplies == true
+    }.map { [$0] }).filter { !$0.isEmpty }
     let group = DispatchGroup()
     let lock = NSLock()
     var candidates: [(VerifiedNostrEvent, PushLeaseCommunity)] = []
@@ -169,9 +172,12 @@ public final class BuzzPushNotificationResolver: BuzzPushNotificationResolving {
   private func queryPolicies(
     _ community: PushLeaseCommunity,
     policies: [PushResolutionPolicy],
+    before: VerifiedNostrEvent? = nil,
+    deadline: Date = Date().addingTimeInterval(8),
     completion: @escaping ((VerifiedNostrEvent, PushLeaseCommunity)?) -> Void
   ) {
-    guard let privateKey = accountPrivateKey(for: community), community.pubkey?.isEmpty == false else {
+    guard deadline.timeIntervalSinceNow > 0,
+      let privateKey = accountPrivateKey(for: community), community.pubkey?.isEmpty == false else {
       completion(nil)
       return
     }
@@ -180,7 +186,14 @@ public final class BuzzPushNotificationResolver: BuzzPushNotificationResolving {
       let relayURL = community.relayURL,
       let url = URL(string: "/query", relativeTo: relayURL),
       let body = try? JSONSerialization.data(
-        withJSONObject: policies.map { $0.filter.queryFilter(since: nil, limit: 10) }
+        withJSONObject: policies.map { policy in
+          var filter = policy.filter.queryFilter(since: nil, limit: 10)
+          if let before {
+            filter["until"] = before.createdAt
+            filter["before_id"] = before.id
+          }
+          return filter
+        }
       )
     else {
       completion(nil)
@@ -189,7 +202,7 @@ public final class BuzzPushNotificationResolver: BuzzPushNotificationResolving {
     var request = URLRequest(url: url)
     request.httpMethod = "POST"
     request.httpBody = body
-    request.timeoutInterval = 8
+    request.timeoutInterval = max(0.001, deadline.timeIntervalSinceNow)
     request.setValue("application/json", forHTTPHeaderField: "Content-Type")
     guard
       let auth = try? NostrHTTPAuth.authorizationHeader(
@@ -225,6 +238,17 @@ public final class BuzzPushNotificationResolver: BuzzPushNotificationResolving {
         },
         community: community
       )
+      if candidate == nil, authoredQuery, events.count == 10,
+        let oldest = events.min(by: {
+          $0.createdAt == $1.createdAt ? $0.id > $1.id : $0.createdAt < $1.createdAt
+        }), oldest.hasValidIDAndSignature(),
+        before.map({ oldest.createdAt < $0.createdAt
+          || (oldest.createdAt == $0.createdAt && oldest.id > $0.id) }) ?? true
+      {
+        self.queryPolicies(community, policies: policies, before: oldest,
+          deadline: deadline, completion: completion)
+        return
+      }
       completion(candidate.map { ($0, community) })
     }.resume()
   }

mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushNotificationResolverTests.swift

@@ -70,6 +70,52 @@ final class BuzzPushNotificationResolverTests: XCTestCase {
     XCTAssertEqual(URLProtocolStub.requests.count, 3)
   }
 
+  func testAuthoredQueryPagesPastMutedAndSelfRepliesAtSameTimestamp() throws {
+    let mutedChannel = "223e4567-e89b-42d3-a456-426614174000"
+    let policy = PushResolutionPolicy(
+      filter: PushLeaseFilter(kinds: [9], authoredThreadReplies: true, since: Self.now - 10),
+      ignore: [PushLeaseFilter(kinds: [9], authors: [Self.ownPubkey]),
+        PushLeaseFilter(kinds: [9], hTags: [mutedChannel])])
+    let snapshot = PushLeaseCommunity(id: "community-id", name: "Community",
+      relayUrl: "https://relay.example", pubkey: Self.ownPubkey,
+      policies: [policy], notificationSessionID: "test-session")
+    let rejected = try (0..<10).map { index in
+      try Self.signedEvent(privateKey: index < 5 ? Self.privateKey : Self.profilePrivateKey,
+        createdAt: Self.now, kind: 9,
+        tags: [["h", index < 5 ? Self.channelID : mutedChannel]], content: "Rejected \(index)")
+    }.sorted { $0.id < $1.id }
+    let eligible = try Self.signedEvent(privateKey: Self.profilePrivateKey,
+      createdAt: Self.now - 1, kind: 9, tags: [["h", Self.channelID]], content: "Eligible older reply")
+    URLProtocolStub.handler = { request in
+      var body = request.httpBody ?? Data()
+      if let stream = request.httpBodyStream {
+        stream.open()
+        defer { stream.close() }
+        var buffer = [UInt8](repeating: 0, count: 4096)
+        while stream.hasBytesAvailable {
+          let count = stream.read(&buffer, maxLength: buffer.count)
+          if count <= 0 { break }
+          body.append(contentsOf: buffer.prefix(count))
+        }
+      }
+      let filters = try XCTUnwrap(JSONSerialization.jsonObject(with: body) as? [[String: Any]])
+      guard let filter = filters.first, filter["authored_thread_replies"] as? Bool == true else {
+        return Self.response(request, status: 200, data: Data("[]".utf8))
+      }
+      XCTAssertEqual(filter["since"] as? Int, Self.now - 10)
+      if filter["before_id"] != nil {
+        XCTAssertEqual(filter["before_id"] as? String, rejected.last?.id)
+        XCTAssertEqual(filter["until"] as? Int, Self.now)
+      }
+      return Self.response(request, status: 200,
+        data: try JSONEncoder().encode(filter["before_id"] == nil ? rejected : [eligible]),
+        headers: ["x-buzz-query-extensions": "buzz-authored-thread-replies-v1"])
+    }
+    let result = resolve(makeResolver(communitiesData: try snapshotData([snapshot])))
+    XCTAssertEqual(result?.body, "Eligible older reply")
+    XCTAssertEqual(URLProtocolStub.requests.count, 3)
+  }
+
   func testAuthoredThreadQueryFailsClosedOnOlderRelay() throws {
     let snapshot = PushLeaseCommunity(id: "community-id", name: "Community",
       relayUrl: "https://relay.example", pubkey: Self.ownPubkey,