From bf90c289e22ff0292f917f56fd2dbc10710eccfe Mon Sep 17 00:00:00 2001 From: npub1tquskdu6yc4h8l7xxtceculxw600grekeq0xg2ukqfrwl7vrzg3quz3gmp <58390b379a262b73ffc632f19c73e6769ef40f36c81e642b960246eff9831222@buzz.block.builderlab.xyz> Date: Tue, 28 Jul 2026 00:17:52 -0700 Subject: [PATCH] fix(mobile/ios): resume bounded push catch-up Page raw relay tails across NSE wakes, retain displayed IDs above the floor, and advance only after complete oldest-first coverage. Co-authored-by: npub1tquskdu6yc4h8l7xxtceculxw600grekeq0xg2ukqfrwl7vrzg3quz3gmp <58390b379a262b73ffc632f19c73e6769ef40f36c81e642b960246eff9831222@buzz.block.builderlab.xyz> Signed-off-by: npub1tquskdu6yc4h8l7xxtceculxw600grekeq0xg2ukqfrwl7vrzg3quz3gmp <58390b379a262b73ffc632f19c73e6769ef40f36c81e642b960246eff9831222@buzz.block.builderlab.xyz> --- .../Sources/BuzzPushKit/PushCatchUp.swift | 181 +++++++++----- .../BuzzPushKit/PushConsumptionState.swift | 76 +++++- .../BuzzPushKitTests/PushCatchUpTests.swift | 114 ++++++--- .../PushConsumptionStateTests.swift | 45 ++++ .../NotificationService.swift | 233 ++++++++++++++---- 5 files changed, 501 insertions(+), 148 deletions(-) diff --git a/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushCatchUp.swift b/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushCatchUp.swift index f3e76fdcb..a67d01679 100644 --- a/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushCatchUp.swift +++ b/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushCatchUp.swift @@ -10,69 +10,136 @@ public struct PushCatchUpSelection: Sendable { } } -public enum PushCatchUp { - /// Build catch-up filters that preserve the lease's constraints. The broad - /// timestamp filter finds newer events, while composite pages cover every gap - /// in the active second's exact delivered-ID set. A separate exact-id filter - /// keeps the last displayed event available only for duplicate cleanup. - public static func queryFilters( +public enum PushCatchUpStopReason: Equatable, Sendable { + case complete + case pageBudgetExceeded + case deadlineExceeded +} + +public struct PushCatchUpPager { + /// The NSE has about eight seconds. Reserve two seconds for state persistence, + /// duplicate absorption, and content delivery after bounded catch-up. + public static let traversalSeconds: TimeInterval = 6 + public static let pageLimit = 10 + public static let maximumPages = 12 + + private struct Query { + let filter: PushLeaseFilter + let hTag: String? + } + + private let queries: [Query] + private let since: Int? + private let deadline: Date + private let pageLimit: Int + private let maximumPages: Int + private var subscriptionIndex: Int + private var rawTail: PushEventPosition? + private var pagesRequested = 0 + + public private(set) var stopReason: PushCatchUpStopReason? + + public init( subscriptions: [PushLeaseSubscription], - cursor: PushEventPosition?, - delivered: [PushEventPosition], - lastDisplayed: PushEventPosition?, - limit: Int - ) -> [[String: Any]] { - subscriptions.flatMap { subscription in - var filters: [[String: Any]] = [] - if let cursor { - // `+ 1` is safe only as the strictly-later half of this pair. The - // active-second filters below start at the bottom and drain same-second - // rows without making event-ID ordering the dedupe authority. - if cursor.createdAt < Int.max { - filters.append( - subscription.filter.queryFilter( - since: cursor.createdAt + 1, - limit: limit - ) - ) - } - - var activeHead = subscription.filter.queryFilter( - since: cursor.createdAt, - limit: limit - ) - activeHead["until"] = cursor.createdAt - filters.append(activeHead) - - let activeDelivered = delivered - .filter { $0.createdAt == cursor.createdAt } - .sorted() - for boundaryIndex in stride( - from: limit - 1, - to: activeDelivered.count, - by: limit - ) { - var page = subscription.filter.queryFilter( - since: cursor.createdAt, - limit: limit - ) - page["until"] = cursor.createdAt - page["before_id"] = activeDelivered[boundaryIndex].id - filters.append(page) - } - } else { - filters.append(subscription.filter.queryFilter(since: nil, limit: limit)) + since: Int?, + scan: PushCatchUpScan = PushCatchUpScan(), + startedAt: Date = Date(), + traversalSeconds: TimeInterval = Self.traversalSeconds, + pageLimit: Int = Self.pageLimit, + maximumPages: Int = Self.maximumPages + ) { + precondition(traversalSeconds > 0, "Push catch-up traversal allowance must be positive") + precondition(pageLimit > 0, "Push catch-up page limit must be positive") + precondition(maximumPages > 0, "Push catch-up page budget must be positive") + queries = subscriptions.flatMap { subscription in + guard let hTags = subscription.filter.hTags, hTags.count > 1 else { + return [Query(filter: subscription.filter, hTag: nil)] } + return hTags.map { Query(filter: subscription.filter, hTag: $0) } + } + self.since = since + deadline = startedAt.addingTimeInterval(traversalSeconds) + self.pageLimit = pageLimit + self.maximumPages = maximumPages - if let lastDisplayed { - var duplicateFallback = subscription.filter.queryFilter(since: nil, limit: 1) - duplicateFallback["ids"] = [lastDisplayed.id] - filters.append(duplicateFallback) - } - return filters + // A subscription snapshot can legitimately shrink between NSE wakes. Treat + // a scan beyond the new query set as exhausted so the caller clears it and + // begins a fresh pass on the next wake. + if scan.subscriptionIndex >= queries.count { + subscriptionIndex = queries.count + rawTail = nil + } else { + subscriptionIndex = scan.subscriptionIndex + rawTail = scan.before } } + /// The position to persist for the next wake. A completed traversal clears + /// its scan so newly arrived events are visible from the top of the next pass. + public var scan: PushCatchUpScan { + guard stopReason != .complete else { return PushCatchUpScan() } + return PushCatchUpScan(subscriptionIndex: subscriptionIndex, before: rawTail) + } + + public func remainingTraversalSeconds(now: Date = Date()) -> TimeInterval { + max(0, deadline.timeIntervalSince(now)) + } + + /// Return the next raw relay page. Continuation is derived only from the + /// observed raw page tail, never from post-selection or delivered-ID state. + public mutating func nextFilter(now: Date = Date()) -> [String: Any]? { + guard stopReason == nil else { return nil } + guard subscriptionIndex < queries.count else { + stopReason = .complete + return nil + } + guard pagesRequested < maximumPages else { + stopReason = .pageBudgetExceeded + return nil + } + guard now < deadline else { + stopReason = .deadlineExceeded + return nil + } + + let query = queries[subscriptionIndex] + var filter = query.filter.queryFilter(since: since, limit: pageLimit) + if let hTag = query.hTag { + // The HTTP relay narrows a multi-value #h filter to its first channel. + // One filter per channel avoids depending on that broken contract. + filter["#h"] = [hTag] + } + if let rawTail { + filter["until"] = rawTail.createdAt + filter["before_id"] = rawTail.id + } + pagesRequested += 1 + return filter + } + + public mutating func receive(rawPage: [VerifiedNostrEvent]) { + guard stopReason == nil else { return } + let ordered = rawPage.sorted { + $0.createdAt == $1.createdAt ? $0.id < $1.id : $0.createdAt > $1.createdAt + } + if ordered.count < pageLimit { + // Every emitted filter is fully represented by the relay's SQL query. + // In particular, #h is exact after splitting, so a short raw page proves + // exhaustion rather than reflecting post-LIMIT filtering. + subscriptionIndex += 1 + rawTail = nil + if subscriptionIndex == queries.count { + stopReason = .complete + } + } else { + rawTail = ordered.last.map { + PushEventPosition(createdAt: $0.createdAt, id: $0.id) + } + } + } +} + +public enum PushCatchUp { /// Return selectable events first and consumed duplicate fallbacks last. /// Duplicate cleanup therefore never competes with forward progress. public static func orderedSelections( diff --git a/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushConsumptionState.swift b/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushConsumptionState.swift index b1af07f52..2b8d1eb8a 100644 --- a/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushConsumptionState.swift +++ b/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushConsumptionState.swift @@ -24,19 +24,52 @@ public struct PushEventPosition: Codable, Equatable, Comparable, Sendable { } } +public struct PushCatchUpScan: Codable, Equatable, Sendable { + public let subscriptionIndex: Int + public let before: PushEventPosition? + + public init(subscriptionIndex: Int = 0, before: PushEventPosition? = nil) { + precondition(subscriptionIndex >= 0, "Push catch-up scan index cannot be negative") + self.subscriptionIndex = subscriptionIndex + self.before = before + } + + enum CodingKeys: String, CodingKey { + case subscriptionIndex + case before + } + + public init(from decoder: Decoder) throws { + let container = try decoder.container(keyedBy: CodingKeys.self) + let subscriptionIndex = try container.decode(Int.self, forKey: .subscriptionIndex) + guard subscriptionIndex >= 0 else { + throw DecodingError.dataCorruptedError( + forKey: .subscriptionIndex, + in: container, + debugDescription: "Push catch-up scan index cannot be negative" + ) + } + self.subscriptionIndex = subscriptionIndex + before = try container.decodeIfPresent(PushEventPosition.self, forKey: .before) + } +} + public struct PushOriginState: Codable, Equatable, Sendable { public let cursor: PushEventPosition? public let delivered: [PushEventPosition] public let lastDisplayed: PushEventPosition? + public let scan: PushCatchUpScan public init( cursor: PushEventPosition? = nil, delivered: [PushEventPosition] = [], - lastDisplayed: PushEventPosition? = nil + lastDisplayed: PushEventPosition? = nil, + scan: PushCatchUpScan = PushCatchUpScan() ) { self.cursor = cursor self.delivered = delivered self.lastDisplayed = lastDisplayed + self.scan = scan } public func contains(eventID: String) -> Bool { @@ -44,23 +77,24 @@ public struct PushOriginState: Codable, Equatable, Sendable { } public func consuming( - _ position: PushEventPosition + _ position: PushEventPosition, + advanceFloor: Bool = true, + scan: PushCatchUpScan? = nil ) -> PushOriginState { - let nextCursor = max(cursor ?? position, position) - let movedToNewSecond = cursor.map { position.createdAt > $0.createdAt } ?? true - var byID = Dictionary( - uniqueKeysWithValues: delivered - .filter { !movedToNewSecond || $0.createdAt == position.createdAt } - .map { ($0.id, $0) } - ) + let nextCursor = advanceFloor ? max(cursor ?? position, position) : cursor + var byID = Dictionary(uniqueKeysWithValues: delivered.map { ($0.id, $0) }) byID[position.id] = position let retained = byID.values - .filter { $0.createdAt >= nextCursor.createdAt } + .filter { displayedPosition in + guard let nextCursor else { return true } + return displayedPosition.createdAt >= nextCursor.createdAt + } .sorted() return PushOriginState( cursor: nextCursor, delivered: Array(retained), - lastDisplayed: position + lastDisplayed: position, + scan: scan ?? self.scan ) } } @@ -130,9 +164,25 @@ public struct PushConsumptionState: Codable, Equatable, Sendable { public mutating func consume( _ position: PushEventPosition, - for origin: String + for origin: String, + advanceFloor: Bool = true, + scan: PushCatchUpScan? = nil ) { - origins[origin] = state(for: origin).consuming(position) + origins[origin] = state(for: origin).consuming( + position, + advanceFloor: advanceFloor, + scan: scan + ) + } + + public mutating func updateScan(_ scan: PushCatchUpScan, for origin: String) { + let originState = state(for: origin) + origins[origin] = PushOriginState( + cursor: originState.cursor, + delivered: originState.delivered, + lastDisplayed: originState.lastDisplayed, + scan: scan + ) } public mutating func removeInactiveOrigins(_ activeOrigins: Set) { diff --git a/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushCatchUpTests.swift b/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushCatchUpTests.swift index 870d4f3ca..32a3d89cf 100644 --- a/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushCatchUpTests.swift +++ b/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushCatchUpTests.swift @@ -96,14 +96,18 @@ final class PushCatchUpTests: XCTestCase { state.consume(PushEventPosition(createdAt: 1_000, id: higherID), for: "origin") let originState = state.state(for: "origin") - let filters = PushCatchUp.queryFilters( + var pager = PushCatchUpPager( subscriptions: [subscription], - cursor: originState.cursor, - delivered: originState.delivered, - lastDisplayed: originState.lastDisplayed, - limit: 10 + since: state.querySince(for: "origin", now: 2_000), + scan: originState.scan, + pageLimit: 10 ) - let relayPage = relayResponse(events: [lateLow, high], filters: filters) + var relayPage: [VerifiedNostrEvent] = [] + while let filter = pager.nextFilter() { + let page = relayResponse(events: [lateLow, high], filters: [filter]) + relayPage.append(contentsOf: page) + pager.receive(rawPage: page) + } let selections = PushCatchUp.orderedSelections( events: relayPage, origin: "origin", @@ -116,41 +120,79 @@ final class PushCatchUpTests: XCTestCase { XCTAssertEqual(selections.map(\.wasPreviouslyConsumed), [false, true]) } - func testCompositeSameSecondQueryReachesSuccessorsBeyondFirstRelayPage() { + func testRawTailPagingReachesEverySameSecondEvent() { let subscription = self.subscription() let events = (0..<25).map { event(id: String(format: "%064x", $0)) } - var state = PushConsumptionState() - var selected: [String] = [] + var pager = PushCatchUpPager( + subscriptions: [subscription], + since: nil, + pageLimit: 10, + maximumPages: 10 + ) + var observed: Set = [] - while selected.count < events.count { - let originState = state.state(for: "origin") - let filters = PushCatchUp.queryFilters( - subscriptions: [subscription], - cursor: originState.cursor, - delivered: originState.delivered, - lastDisplayed: originState.lastDisplayed, - limit: 10 - ) - let relayPage = relayResponse(events: events, filters: filters) - let selection = PushCatchUp.orderedSelections( - events: relayPage, - origin: "origin", - subscriptions: [subscription], - consumptionState: state, - verify: { _ in true } - ).first - let winner = try! XCTUnwrap(selection) - - XCTAssertFalse(winner.wasPreviouslyConsumed) - XCTAssertFalse(selected.contains(winner.event.id)) - selected.append(winner.event.id) - state.consume( - PushEventPosition(createdAt: winner.event.createdAt, id: winner.event.id), - for: "origin" - ) + while let filter = pager.nextFilter() { + let page = relayResponse(events: events, filters: [filter]) + observed.formUnion(page.map(\.id)) + pager.receive(rawPage: page) } - XCTAssertEqual(selected, events.map(\.id)) + XCTAssertEqual(observed, Set(events.map(\.id))) + XCTAssertEqual(pager.stopReason, .complete) + } + + func testBudgetedTraversalResumesFromPersistedRawTailAndClearsOnlyOnComplete() { + let subscription = self.subscription() + let events = (0..<25).map { + event(id: String(format: "%064x", $0), createdAt: 1_000 + $0) + } + var scan = PushCatchUpScan() + var observations: [String: Int] = [:] + var stops: [PushCatchUpStopReason] = [] + + for _ in 0..<3 { + var pager = PushCatchUpPager( + subscriptions: [subscription], + since: nil, + scan: scan, + pageLimit: 5, + maximumPages: 2 + ) + while let filter = pager.nextFilter() { + let page = relayResponse(events: events, filters: [filter]) + for event in page { observations[event.id, default: 0] += 1 } + pager.receive(rawPage: page) + } + stops.append(try! XCTUnwrap(pager.stopReason)) + scan = pager.scan + } + + XCTAssertEqual(stops, [.pageBudgetExceeded, .pageBudgetExceeded, .complete]) + XCTAssertEqual(observations.count, events.count) + XCTAssertTrue(observations.values.allSatisfy { $0 == 1 }) + XCTAssertEqual(scan, PushCatchUpScan()) + } + + func testMultiChannelSubscriptionEmitsOneHTTPFilterPerHTag() { + let firstChannel = "11111111-1111-4111-8111-111111111111" + let secondChannel = "22222222-2222-4222-8222-222222222222" + let subscription = PushLeaseSubscription( + filter: PushLeaseFilter( + kinds: [9], + hTags: [firstChannel, secondChannel] + ), + notificationClass: "default" + ) + var pager = PushCatchUpPager(subscriptions: [subscription], since: nil) + var emittedHTags: [[String]] = [] + + while let filter = pager.nextFilter() { + emittedHTags.append(filter["#h"] as? [String] ?? []) + pager.receive(rawPage: []) + } + + XCTAssertEqual(emittedHTags, [[firstChannel], [secondChannel]]) + XCTAssertEqual(pager.stopReason, .complete) } private func subscription() -> PushLeaseSubscription { diff --git a/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushConsumptionStateTests.swift b/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushConsumptionStateTests.swift index 065abe166..56e41771b 100644 --- a/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushConsumptionStateTests.swift +++ b/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushConsumptionStateTests.swift @@ -62,6 +62,51 @@ final class PushConsumptionStateTests: XCTestCase { XCTAssertEqual(state.state(for: "origin").delivered, [nextSecond]) } + func testDisplayedIDsAtOrAboveFloorAreRetained() { + var state = PushConsumptionState() + let floor = PushEventPosition(createdAt: 1_000, id: "floor") + let newer = PushEventPosition(createdAt: 1_002, id: "newer") + let middle = PushEventPosition(createdAt: 1_001, id: "middle") + + state.consume(floor, for: "origin") + state.consume(newer, for: "origin", advanceFloor: false) + state.consume(middle, for: "origin", advanceFloor: false) + + let originState = state.state(for: "origin") + XCTAssertEqual(originState.cursor, floor) + XCTAssertEqual(originState.delivered, [floor, middle, newer]) + } + + func testIncompleteTraversalKeepsFloorAndPersistsScanThroughConsumeAndCodable() throws { + let floor = PushEventPosition(createdAt: 1_000, id: "floor") + let displayed = PushEventPosition(createdAt: 1_015, id: "displayed") + let before = PushEventPosition(createdAt: 1_010, id: "raw-tail") + let scan = PushCatchUpScan(subscriptionIndex: 2, before: before) + var state = PushConsumptionState() + state.consume(floor, for: "origin") + + state.consume(displayed, for: "origin", advanceFloor: false, scan: scan) + + let originState = state.state(for: "origin") + XCTAssertEqual(originState.cursor, floor) + XCTAssertEqual(originState.scan, scan) + XCTAssertTrue(state.hasConsumed(eventID: displayed.id, for: "origin")) + let decoded = try JSONDecoder().decode( + PushConsumptionState.self, + from: JSONEncoder().encode(state) + ) + XCTAssertEqual(decoded.state(for: "origin"), originState) + } + + func testNegativePersistedScanIndexIsRejected() throws { + let data = try XCTUnwrap( + #"{"version":1,"origins":{"origin":{"delivered":[],"scan":{"subscriptionIndex":-1}}}}"# + .data(using: .utf8) + ) + + XCTAssertThrowsError(try JSONDecoder().decode(PushConsumptionState.self, from: data)) + } + func testLowerIDSiblingArrivingAfterHigherIDRemainsSelectable() { var state = PushConsumptionState() let higher = PushEventPosition(createdAt: 1_000, id: "ff") diff --git a/mobile/ios/NotificationService/NotificationService.swift b/mobile/ios/NotificationService/NotificationService.swift index f89869df0..b90fa7698 100644 --- a/mobile/ios/NotificationService/NotificationService.swift +++ b/mobile/ios/NotificationService/NotificationService.swift @@ -97,6 +97,8 @@ final class BuzzPushNotificationResolver: BuzzPushNotificationResolving { let event: VerifiedNostrEvent let community: PushLeaseCommunity let wasPreviouslyConsumed: Bool + let catchUpStopReason: PushCatchUpStopReason + let catchUpScan: PushCatchUpScan } private let session: URLSession @@ -151,34 +153,76 @@ final class BuzzPushNotificationResolver: BuzzPushNotificationResolving { let group = DispatchGroup() let lock = NSLock() - var candidates: [Candidate] = [] - var diagnostics: [String] = [] + var outcomes: [String: QueryResult] = [:] for community in communities { group.enter() query(community, consumptionState: consumptionState) { result in lock.lock() - switch result { - case .candidate(let candidate): candidates.append(candidate) - case .diagnostic(let diagnostic): diagnostics.append(diagnostic) - case .none: break - } + outcomes[community.id] = result lock.unlock() group.leave() } } group.notify(queue: .global(qos: .userInitiated)) { [weak self] in guard let self else { return } + let candidates = outcomes.values.compactMap { result -> Candidate? in + guard case .candidate(let candidate) = result else { return nil } + return candidate + } + let diagnostics = outcomes.values.compactMap { result -> String? in + switch result { + case .diagnostic(let diagnostic): return diagnostic + case .traversal(let traversal): return traversal.diagnostic + case .candidate(let candidate): + return candidate.catchUpStopReason == .complete + ? nil + : Self.incompleteTraversalMessage + } + } + let traversals = outcomes.values.compactMap { result -> CommunityTraversal? in + switch result { + case .candidate(let candidate): + return CommunityTraversal( + community: candidate.community, + scan: candidate.catchUpScan, + diagnostic: nil + ) + case .traversal(let traversal): return traversal + case .diagnostic: return nil + } + } let sorted = candidates.sorted { lhs, rhs in if lhs.wasPreviouslyConsumed != rhs.wasPreviouslyConsumed { return !lhs.wasPreviouslyConsumed } if lhs.event.createdAt != rhs.event.createdAt { - return lhs.event.createdAt > rhs.event.createdAt + return lhs.event.createdAt < rhs.event.createdAt } if lhs.event.id != rhs.event.id { return lhs.event.id < rhs.event.id } return lhs.community.id < rhs.community.id } + let incompleteTraversal = outcomes.values.contains { result in + switch result { + case .candidate(let candidate): + return candidate.catchUpStopReason != .complete + case .traversal(let traversal): + return traversal.diagnostic == Self.incompleteTraversalMessage + case .diagnostic: + return false + } + } guard let winner = sorted.first else { + do { + try loaded.store.update { state in + state.removeInactiveOrigins(Set(communities.map(\.id))) + for traversal in traversals { + state.updateScan(traversal.scan, for: traversal.community.id) + } + } + } catch { + completion(.diagnostic("Buzz could not save notification history.")) + return + } completion(diagnostics.sorted().first.map(BuzzPushResolutionResult.diagnostic) ?? .none) return } @@ -191,12 +235,22 @@ final class BuzzPushNotificationResolver: BuzzPushNotificationResolving { do { try loaded.store.update { state in state.removeInactiveOrigins(Set(communities.map(\.id))) + for traversal in traversals { + guard traversal.community.id != winner.community.id else { continue } + state.updateScan(traversal.scan, for: traversal.community.id) + } if state.hasConsumed(eventID: position.id, for: winner.community.id) { + state.updateScan(winner.catchUpScan, for: winner.community.id) shouldPresent = true return } guard state.canSelect(position, for: winner.community.id) else { return } - state.consume(position, for: winner.community.id) + state.consume( + position, + for: winner.community.id, + advanceFloor: winner.catchUpStopReason == .complete, + scan: winner.catchUpScan + ) shouldPresent = true } } catch { @@ -207,16 +261,29 @@ final class BuzzPushNotificationResolver: BuzzPushNotificationResolving { completion(.none) return } + guard !incompleteTraversal else { + completion(.diagnostic(Self.incompleteTraversalMessage)) + return + } absorbSequentialDuplicate(of: winner.resolution, completion: completion) } } private enum QueryResult { case candidate(Candidate) + case traversal(CommunityTraversal) case diagnostic(String) - case none } + private struct CommunityTraversal { + let community: PushLeaseCommunity + let scan: PushCatchUpScan + let diagnostic: String? + } + + private static let incompleteTraversalMessage = + "Open Buzz to finish checking for new activity." + private func query( _ community: PushLeaseCommunity, consumptionState: PushConsumptionState, @@ -234,27 +301,101 @@ final class BuzzPushNotificationResolver: BuzzPushNotificationResolving { return } let originState = consumptionState.state(for: community.id) - let filters = PushCatchUp.queryFilters( + let pager = PushCatchUpPager( subscriptions: subscriptions, - cursor: originState.cursor, - delivered: originState.delivered, - lastDisplayed: originState.lastDisplayed, - limit: 10 + since: consumptionState.querySince(for: community.id), + scan: originState.scan ) - guard let body = try? JSONSerialization.data(withJSONObject: filters) else { - completion(.diagnostic("Buzz notification subscriptions are invalid.")) - return - } guard let relayURL = community.relayURL, let url = URL(string: "/query", relativeTo: relayURL) else { completion(.diagnostic("Buzz notification relay URL is invalid.")) return } + + queryNextPage( + pager: pager, + eventsByID: [:], + url: url, + privateKey: privateKey + ) { result in + switch result { + case .success(let traversalResult): + let traversal = CommunityTraversal( + community: community, + scan: traversalResult.scan, + diagnostic: traversalResult.stopReason == .complete + ? nil + : Self.incompleteTraversalMessage + ) + let candidate = Self.decodeCandidate( + events: Array(traversalResult.eventsByID.values), + community: community, + subscriptions: subscriptions, + consumptionState: consumptionState, + stopReason: traversalResult.stopReason, + scan: traversalResult.scan + ) + completion(candidate.map(QueryResult.candidate) ?? .traversal(traversal)) + case .failure: + completion( + .traversal( + CommunityTraversal( + community: community, + scan: originState.scan, + diagnostic: nil + ) + ) + ) + } + } + } + + private enum CatchUpQueryError: Error { + case invalidFilters + case authentication + case relay + } + + private struct CatchUpTraversal { + let eventsByID: [String: VerifiedNostrEvent] + let stopReason: PushCatchUpStopReason + let scan: PushCatchUpScan + } + + private func queryNextPage( + pager: PushCatchUpPager, + eventsByID: [String: VerifiedNostrEvent], + url: URL, + privateKey: String, + completion: @escaping (Result) -> Void + ) { + var pager = pager + guard let filter = pager.nextFilter() else { + guard let stopReason = pager.stopReason else { + completion(.failure(.relay)) + return + } + completion( + .success( + CatchUpTraversal( + eventsByID: eventsByID, + stopReason: stopReason, + scan: pager.scan + ) + ) + ) + return + } + guard let body = try? JSONSerialization.data(withJSONObject: [filter]) else { + completion(.failure(.invalidFilters)) + return + } + var request = URLRequest(url: url) request.httpMethod = "POST" request.httpBody = body - request.timeoutInterval = 8 + request.timeoutInterval = max(1, pager.remainingTraversalSeconds()) request.setValue("application/json", forHTTPHeaderField: "Content-Type") guard let auth = try? NostrHTTPAuth.authorizationHeader( @@ -264,36 +405,44 @@ final class BuzzPushNotificationResolver: BuzzPushNotificationResolving { privateKeyHex: privateKey ) else { - completion(.diagnostic("Buzz could not authenticate notification catch-up.")) + completion(.failure(.authentication)) return } request.setValue(auth, forHTTPHeaderField: "Authorization") - session.dataTask(with: request) { data, response, _ in - guard let response = response as? HTTPURLResponse, + + session.dataTask(with: request) { [weak self] data, response, _ in + guard let self, + let response = response as? HTTPURLResponse, (200..<300).contains(response.statusCode), let data, let events = try? JSONDecoder().decode([VerifiedNostrEvent].self, from: data) else { - completion(.none) + completion(.failure(.relay)) return } - completion( - Self.decodeResolution( - events: events, - community: community, - subscriptions: subscriptions, - consumptionState: consumptionState - ) + var nextEventsByID = eventsByID + for event in events { + nextEventsByID[event.id] = event + } + pager.receive(rawPage: events) + self.queryNextPage( + pager: pager, + eventsByID: nextEventsByID, + url: url, + privateKey: privateKey, + completion: completion ) }.resume() } - private static func decodeResolution( + private static func decodeCandidate( events: [VerifiedNostrEvent], community: PushLeaseCommunity, subscriptions: [PushLeaseSubscription], - consumptionState: PushConsumptionState - ) -> QueryResult { + consumptionState: PushConsumptionState, + stopReason: PushCatchUpStopReason, + scan: PushCatchUpScan + ) -> Candidate? { let matching = PushCatchUp.orderedSelections( events: events, origin: community.id, @@ -329,16 +478,16 @@ final class BuzzPushNotificationResolver: BuzzPushNotificationResolving { identity: identity ) } - return .candidate( - Candidate( - resolution: resolution, - event: event, - community: community, - wasPreviouslyConsumed: selection.wasPreviouslyConsumed - ) + return Candidate( + resolution: resolution, + event: event, + community: community, + wasPreviouslyConsumed: selection.wasPreviouslyConsumed, + catchUpStopReason: stopReason, + catchUpScan: scan ) } - return .none + return nil } private func absorbSequentialDuplicate(