Chunk stateless feed requests to support more than 500 subscriptions (#967)

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..<Swift.min($0 + size, count)])
+        }
+    }
+}

Yattee/Services/API/YatteeServerAPI.swift

@@ -349,18 +349,32 @@ actor YatteeServerAPI: InstanceAPI {
 
     // MARK: - Stateless Feed Endpoints
 
-    /// Fetches feed using stateless POST endpoint with channel list.
+    /// Maximum number of channels per stateless feed request. The server rejects larger channel
+    /// lists with HTTP 422 (`max_length=500` on its feed and feed-status request models).
+    static let maxChannelsPerFeedRequest = 500
+
+    /// Fetches feed using stateless POST endpoint with channel list. Channel lists above the
+    /// server's per-request limit are sent in chunks and the responses merged into one.
     func postFeed(channels: [StatelessChannelRequest], limit: Int, offset: Int, instance: Instance) async throws -> StatelessFeedResponse {
-        let body = StatelessFeedRequest(channels: channels, limit: limit, offset: offset)
+        var responses: [StatelessFeedResponse] = []
-        let endpoint = GenericEndpoint.post("/api/v1/feed", body: body)
+        for chunk in channels.chunked(into: Self.maxChannelsPerFeedRequest) {
-        return try await httpClient.fetch(endpoint, baseURL: instance.url)
+            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)
+        var responses: [StatelessFeedStatusResponse] = []
-        let endpoint = GenericEndpoint.post("/api/v1/feed/status", body: body)
+        for chunk in channels.chunked(into: Self.maxChannelsPerFeedRequest) {
-        return try await httpClient.fetch(endpoint, baseURL: instance.url)
+            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

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..<Swift.min($0 + size, count)])
-        }
-    }
-}

YatteeTests/YatteeServerFeedChunkingTests.swift

@@ -0,0 +1,153 @@
+//
+//  YatteeServerFeedChunkingTests.swift
+//  YatteeTests
+//
+//  Tests for chunking and merging of stateless feed requests against Yattee Server,
+//  which rejects more than 500 channels per request.
+//
+
+import Testing
+import Foundation
+@testable import Yattee
+
+@Suite("Yattee Server Feed Chunking Tests")
+struct YatteeServerFeedChunkingTests {
+
+    private func video(id: String, published: Int64?) -> 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)
+    }
+}