From d46002e99ac086e9ea90d3eb431be3dbf9b8ba75 Mon Sep 17 00:00:00 2001 From: Phil-Bastian Berndt Date: Sun, 23 Aug 2026 12:22:07 +0200 Subject: [PATCH] Chunk stateless feed requests to support more than 500 subscriptions (#967) --- Yattee/Extensions/Array+Chunked.swift | 17 ++ Yattee/Services/API/YatteeServerAPI.swift | 75 ++++++++- Yattee/Services/DeArrowBrandingProvider.swift | 11 -- .../YatteeServerFeedChunkingTests.swift | 153 ++++++++++++++++++ 4 files changed, 238 insertions(+), 18 deletions(-) create mode 100644 Yattee/Extensions/Array+Chunked.swift create mode 100644 YatteeTests/YatteeServerFeedChunkingTests.swift diff --git a/Yattee/Extensions/Array+Chunked.swift b/Yattee/Extensions/Array+Chunked.swift new file mode 100644 index 00000000..3758af4b --- /dev/null +++ b/Yattee/Extensions/Array+Chunked.swift @@ -0,0 +1,17 @@ +// +// Array+Chunked.swift +// Yattee +// +// Splitting arrays into fixed-size chunks. +// + +import Foundation + +extension Array { + /// Splits the array into chunks of the specified size. + func chunked(into size: Int) -> [[Element]] { + stride(from: 0, to: count, by: size).map { + Array(self[$0.. StatelessFeedResponse { - let body = StatelessFeedRequest(channels: channels, limit: limit, offset: offset) - let endpoint = GenericEndpoint.post("/api/v1/feed", body: body) - return try await httpClient.fetch(endpoint, baseURL: instance.url) + var responses: [StatelessFeedResponse] = [] + for chunk in channels.chunked(into: Self.maxChannelsPerFeedRequest) { + let body = StatelessFeedRequest(channels: chunk, limit: limit, offset: offset) + let endpoint = GenericEndpoint.post("/api/v1/feed", body: body) + responses.append(try await httpClient.fetch(endpoint, baseURL: instance.url)) + } + return StatelessFeedResponse.merged(responses, limit: limit) } /// Checks feed status for given channels (lightweight polling). + /// Chunked like `postFeed(channels:limit:offset:instance:)`. func postFeedStatus(channels: [StatelessChannelStatusRequest], instance: Instance) async throws -> StatelessFeedStatusResponse { - let body = StatelessFeedStatusRequest(channels: channels) - let endpoint = GenericEndpoint.post("/api/v1/feed/status", body: body) - return try await httpClient.fetch(endpoint, baseURL: instance.url) + var responses: [StatelessFeedStatusResponse] = [] + for chunk in channels.chunked(into: Self.maxChannelsPerFeedRequest) { + let body = StatelessFeedStatusRequest(channels: chunk) + let endpoint = GenericEndpoint.post("/api/v1/feed/status", body: body) + responses.append(try await httpClient.fetch(endpoint, baseURL: instance.url)) + } + return StatelessFeedStatusResponse.merged(responses) } // MARK: - Channel Metadata @@ -497,6 +511,34 @@ struct StatelessFeedResponse: Decodable, Sendable { func toVideos() -> [Video] { videos.compactMap { $0.toVideo() } } + + /// Merges chunked feed responses into a single response. A single response is returned + /// unchanged so requests within the server's per-request channel limit behave as before. + static func merged(_ responses: [StatelessFeedResponse], limit: Int) -> StatelessFeedResponse { + guard responses.count != 1 else { return responses[0] } + + let videos = responses.flatMap(\.videos) + .sorted { ($0.published ?? 0) > ($1.published ?? 0) } + let limited = Array(videos.prefix(limit)) + + return StatelessFeedResponse( + status: responses.first(where: { !$0.isReady })?.status ?? "ready", + videos: limited, + total: responses.reduce(0) { $0 + $1.total }, + hasMore: responses.contains(where: \.hasMore) || videos.count > limited.count, + readyCount: sumIfAnyPresent(responses.map(\.readyCount)), + pendingCount: sumIfAnyPresent(responses.map(\.pendingCount)), + errorCount: sumIfAnyPresent(responses.map(\.errorCount)), + etaSeconds: responses.compactMap(\.etaSeconds).max() + ) + } + + /// Sums optional counts, staying `nil` when no chunk reported a value (older server + /// versions omit the count fields entirely). + private static func sumIfAnyPresent(_ values: [Int?]) -> Int? { + let present = values.compactMap { $0 } + return present.isEmpty ? nil : present.reduce(0, +) + } } /// Response from stateless feed status endpoint. @@ -523,6 +565,25 @@ struct StatelessFeedStatusResponse: Decodable, Sendable { case pendingCount case errorCount } + + init(status: String, readyCount: Int, pendingCount: Int, errorCount: Int) { + self.status = status + self.readyCount = readyCount + self.pendingCount = pendingCount + self.errorCount = errorCount + } + + /// Merges chunked status responses into a single response. A single response is returned unchanged. + static func merged(_ responses: [StatelessFeedStatusResponse]) -> StatelessFeedStatusResponse { + guard responses.count != 1 else { return responses[0] } + + return StatelessFeedStatusResponse( + status: responses.first(where: { !$0.isReady })?.status ?? "ready", + readyCount: responses.reduce(0) { $0 + $1.readyCount }, + pendingCount: responses.reduce(0) { $0 + $1.pendingCount }, + errorCount: responses.reduce(0) { $0 + $1.errorCount } + ) + } } // MARK: - Channel Metadata Models diff --git a/Yattee/Services/DeArrowBrandingProvider.swift b/Yattee/Services/DeArrowBrandingProvider.swift index b10a25cc..9a9b7b49 100644 --- a/Yattee/Services/DeArrowBrandingProvider.swift +++ b/Yattee/Services/DeArrowBrandingProvider.swift @@ -225,14 +225,3 @@ final class DeArrowBrandingProvider { } } } - -// MARK: - Array Extension - -private extension Array { - /// Splits the array into chunks of the specified size. - func chunked(into size: Int) -> [[Element]] { - stride(from: 0, to: count, by: size).map { - Array(self[$0.. ServerFeedVideo { + ServerFeedVideo( + type: "video", + videoId: id, + title: "Video \(id)", + author: "Author", + authorId: "UC\(id)", + lengthSeconds: 60, + published: published, + publishedText: nil, + viewCount: nil, + videoThumbnails: nil, + extractor: "youtube", + videoUrl: nil, + isUpcoming: nil, + premiereTimestamp: nil + ) + } + + // MARK: - Chunking + + @Test("Chunked splits arrays at the requested size") + func chunkedSplitsAtSize() { + let channels = Array(0..<1018) + let chunks = channels.chunked(into: 500) + #expect(chunks.map(\.count) == [500, 500, 18]) + #expect(chunks.flatMap { $0 } == channels) + } + + @Test("Chunked keeps lists within the size as a single chunk") + func chunkedSingleChunk() { + #expect(Array(0..<500).chunked(into: 500).count == 1) + #expect(Array(0..<3).chunked(into: 500).map(\.count) == [3]) + #expect([Int]().chunked(into: 500).isEmpty) + } + + // MARK: - Feed Response Merging + + @Test("Merging a single feed response returns it unchanged") + func mergedSingleFeedResponsePassthrough() { + // Deliberately unsorted: a single response must not be re-sorted or truncated. + let response = StatelessFeedResponse( + status: "pending", + videos: [video(id: "a", published: 100), video(id: "b", published: 300)], + total: 2, + hasMore: true, + readyCount: 1, + pendingCount: 2, + errorCount: nil, + etaSeconds: 30 + ) + let merged = StatelessFeedResponse.merged([response], limit: 1) + #expect(merged.videos.map(\.videoId) == ["a", "b"]) + #expect(merged.status == "pending") + #expect(merged.total == 2) + } + + @Test("Merging feed responses sorts videos by published date and applies the limit") + func mergedFeedResponsesSortsAndLimits() { + let first = StatelessFeedResponse( + status: "ready", + videos: [video(id: "old", published: 100), video(id: "newest", published: 900)], + total: 2, + hasMore: false, + readyCount: 500, + pendingCount: 0, + errorCount: 0, + etaSeconds: nil + ) + let second = StatelessFeedResponse( + status: "ready", + videos: [video(id: "middle", published: 500), video(id: "undated", published: nil)], + total: 2, + hasMore: false, + readyCount: 18, + pendingCount: 0, + errorCount: 0, + etaSeconds: nil + ) + + let merged = StatelessFeedResponse.merged([first, second], limit: 3) + #expect(merged.videos.map(\.videoId) == ["newest", "middle", "old"]) + #expect(merged.total == 4) + #expect(merged.hasMore) // truncated from 4 to 3 + #expect(merged.readyCount == 518) + #expect(merged.status == "ready") + #expect(merged.isReady) + } + + @Test("Merged feed response is pending while any chunk is pending") + func mergedFeedResponsePendingWhileAnyChunkPending() { + let ready = StatelessFeedResponse( + status: "ready", videos: [], total: 0, hasMore: false, + readyCount: 500, pendingCount: 0, errorCount: 0, etaSeconds: nil + ) + let pending = StatelessFeedResponse( + status: "pending", videos: [], total: 0, hasMore: false, + readyCount: 10, pendingCount: 8, errorCount: 0, etaSeconds: 45 + ) + + let merged = StatelessFeedResponse.merged([ready, pending], limit: 100) + #expect(merged.status == "pending") + #expect(!merged.isReady) + #expect(merged.pendingCount == 8) + #expect(merged.etaSeconds == 45) + } + + @Test("Merged feed response keeps counts nil when no chunk reports them") + func mergedFeedResponseKeepsNilCounts() { + let first = StatelessFeedResponse( + status: "ready", videos: [], total: 0, hasMore: false, + readyCount: nil, pendingCount: nil, errorCount: nil, etaSeconds: nil + ) + let merged = StatelessFeedResponse.merged([first, first], limit: 100) + #expect(merged.readyCount == nil) + #expect(merged.pendingCount == nil) + #expect(merged.errorCount == nil) + #expect(merged.etaSeconds == nil) + } + + // MARK: - Feed Status Merging + + @Test("Merging status responses sums counts and stays pending until all chunks are ready") + func mergedStatusResponses() { + let ready = StatelessFeedStatusResponse(status: "ready", readyCount: 500, pendingCount: 0, errorCount: 1) + let pending = StatelessFeedStatusResponse(status: "pending", readyCount: 10, pendingCount: 8, errorCount: 0) + + let merged = StatelessFeedStatusResponse.merged([ready, pending]) + #expect(merged.status == "pending") + #expect(merged.readyCount == 510) + #expect(merged.pendingCount == 8) + #expect(merged.errorCount == 1) + + let allReady = StatelessFeedStatusResponse.merged([ready, ready]) + #expect(allReady.isReady) + #expect(allReady.readyCount == 1000) + } +}