[Local Catalog] Create full sync service that syncs catalog with batches of paginated products/variations requests (#16071)

jaclync · web-flow · commit c8b64f56ec90 · 2025-09-03T09:22:43.000+08:00
diff --git a/Modules/Sources/Networking/Remote/POSCatalogSyncRemote.swift b/Modules/Sources/Networking/Remote/POSCatalogSyncRemote.swift
@@ -1,8 +1,49 @@
 import Foundation
 
+/// Protocol for POS Catalog Sync Remote operations.
+public protocol POSCatalogSyncRemoteProtocol {
+    /// Loads POS products modified after the specified date for incremental sync.
+    ///
+    /// - Parameters:
+    ///   - modifiedAfter: Only products modified after this date will be returned.
+    ///   - siteID: Site ID to load products from.
+    ///   - pageNumber: Page number for pagination.
+    /// - Returns: Paginated list of POS products.
+    // TODO - remove the periphery ignore comment when the incremental sync is integrated with POS.
+    // periphery:ignore
+    func loadProducts(modifiedAfter: Date, siteID: Int64, pageNumber: Int) async throws -> PagedItems<POSProduct>
+
+    /// Loads POS product variations modified after the specified date for incremental sync.
+    ///
+    /// - Parameters:
+    ///   - modifiedAfter: Only variations modified after this date will be returned.
+    ///   - siteID: Site ID to load variations from.
+    ///   - pageNumber: Page number for pagination.
+    /// - Returns: Paginated list of POS product variations.
+    // TODO - remove the periphery ignore comment when the incremental sync is integrated with POS.
+    // periphery:ignore
+    func loadProductVariations(modifiedAfter: Date, siteID: Int64, pageNumber: Int) async throws -> PagedItems<POSProductVariation>
+
+    /// Loads POS products for full sync.
+    ///
+    /// - Parameters:
+    ///   - siteID: Site ID to load products from.
+    ///   - pageNumber: Page number for pagination.
+    /// - Returns: Paginated list of POS products.
+    func loadProducts(siteID: Int64, pageNumber: Int) async throws -> PagedItems<POSProduct>
+
+    /// Loads POS product variations for full sync.
+    ///
+    /// - Parameters:
+    ///   - siteID: Site ID to load variations from.
+    ///   - pageNumber: Page number for pagination.
+    /// - Returns: Paginated list of POS product variations.
+    func loadProductVariations(siteID: Int64, pageNumber: Int) async throws -> PagedItems<POSProductVariation>
+}
+
 /// POS Catalog Sync: Remote Endpoints
 ///
-public class POSCatalogSyncRemote: Remote {
+public class POSCatalogSyncRemote: Remote, POSCatalogSyncRemoteProtocol {
     private let dateFormatter = ISO8601DateFormatter()
 
     // MARK: - Incremental Sync Endpoints
diff --git a/Modules/Sources/NetworkingCore/Network/AlamofireNetwork.swift b/Modules/Sources/NetworkingCore/Network/AlamofireNetwork.swift
@@ -24,6 +24,7 @@ extension Alamofire.MultipartFormData: MultipartFormData {
 /// AlamofireWrapper: Encapsulates all of the Alamofire OP's
 ///
 public class AlamofireNetwork: Network {
+    /// Lazy-initialized session manager. Use ensuresSessionManagerIsInitialized=true to avoid race conditions with concurrent requests.
     private lazy var alamofireSession: Alamofire.Session = {
         let sessionConfiguration = URLSessionConfiguration.default
         let sessionManager = makeSession(configuration: sessionConfiguration)
@@ -48,10 +49,16 @@ public class AlamofireNetwork: Network {
 
     /// Public Initializer
     ///
-    ///
+    /// - Parameters:
+    ///   - credentials: Authentication credentials for requests.
+    ///   - selectedSite: Publisher for site selection changes.
+    ///   - sessionManager: Optional pre-configured session manager.
+    ///   - ensuresSessionManagerIsInitialized: If true, the session is always set during initialization immediately to avoid lazy initialization race conditions.
+    ///     Defaults to false for backward compatibility. Set to true when making concurrent requests immediately after initialization.
     public required init(credentials: Credentials?,
                          selectedSite: AnyPublisher<JetpackSite?, Never>? = nil,
-                         sessionManager: Alamofire.Session? = nil) {
+                         sessionManager: Alamofire.Session? = nil,
+                         ensuresSessionManagerIsInitialized: Bool = false) {
         self.credentials = credentials
         self.selectedSite = selectedSite
         self.requestConverter = {
@@ -70,6 +77,8 @@ public class AlamofireNetwork: Network {
         self.requestAuthenticator = RequestProcessor(requestAuthenticator: DefaultRequestAuthenticator(credentials: credentials))
         if let sessionManager {
             self.alamofireSession = sessionManager
+        } else if ensuresSessionManagerIsInitialized {
+            self.alamofireSession = makeSession(configuration: URLSessionConfiguration.default)
         }
     }
 
diff --git a/Modules/Sources/Yosemite/Tools/POS/POSCatalogFullSyncService.swift b/Modules/Sources/Yosemite/Tools/POS/POSCatalogFullSyncService.swift
@@ -0,0 +1,156 @@
+import Foundation
+import protocol Networking.POSCatalogSyncRemoteProtocol
+import class Networking.AlamofireNetwork
+import class Networking.POSCatalogSyncRemote
+import CocoaLumberjackSwift
+
+// TODO - remove the periphery ignore comment when the catalog is integrated with POS.
+// periphery:ignore
+public protocol POSCatalogFullSyncServiceProtocol {
+    /// Starts a full catalog sync process
+    /// - Parameter siteID: The site ID to sync catalog for
+    /// - Returns: The synced catalog containing products and variations
+    func startFullSync(for siteID: Int64) async throws -> POSCatalog
+}
+
+/// POS catalog from full sync.
+// TODO - remove the periphery ignore comment when the catalog is integrated with POS.
+// periphery:ignore
+public struct POSCatalog {
+    public let products: [POSProduct]
+    public let variations: [POSProductVariation]
+}
+
+// TODO - remove the periphery ignore comment when the service is integrated with POS.
+// periphery:ignore
+public final class POSCatalogFullSyncService: POSCatalogFullSyncServiceProtocol {
+    private let syncRemote: POSCatalogSyncRemoteProtocol
+    private let batchSize: Int
+
+    public convenience init?(credentials: Credentials?, batchSize: Int = 2) {
+        guard let credentials else {
+            DDLogError("⛔️ Could not create POSCatalogFullSyncService due missing credentials")
+            return nil
+        }
+        let network = AlamofireNetwork(credentials: credentials, ensuresSessionManagerIsInitialized: true)
+        let syncRemote = POSCatalogSyncRemote(network: network)
+        self.init(syncRemote: syncRemote, batchSize: batchSize)
+    }
+
+    init(syncRemote: POSCatalogSyncRemoteProtocol, batchSize: Int) {
+        self.syncRemote = syncRemote
+        self.batchSize = batchSize
+    }
+
+    // MARK: - Protocol Conformance
+
+    public func startFullSync(for siteID: Int64) async throws -> POSCatalog {
+        DDLogInfo("🔄 Starting full catalog sync for site ID: \(siteID)")
+
+        do {
+            let catalog = try await loadCatalog(for: siteID, syncRemote: syncRemote)
+            DDLogInfo("✅ Loaded \(catalog.products.count) products and \(catalog.variations.count) variations for siteID \(siteID)")
+            return catalog
+        } catch {
+            throw error
+        }
+    }
+}
+
+// MARK: - Remote Loading
+
+private extension POSCatalogFullSyncService {
+    func loadCatalog(for siteID: Int64, syncRemote: POSCatalogSyncRemoteProtocol) async throws -> POSCatalog {
+        // Loads products and variations in batches in parallel.
+        async let productsTask = loadAllProducts(for: siteID, syncRemote: syncRemote)
+        async let variationsTask = loadAllProductVariations(for: siteID, syncRemote: syncRemote)
+
+        let (products, variations) = try await (productsTask, variationsTask)
+        return POSCatalog(products: products, variations: variations)
+    }
+
+    func loadAllProducts(for siteID: Int64, syncRemote: POSCatalogSyncRemoteProtocol) async throws -> [POSProduct] {
+        DDLogInfo("🔄 Starting products sync for site ID: \(siteID)")
+
+        var allProducts: [POSProduct] = []
+        var currentPage = 1
+        var hasMorePages = true
+
+        while hasMorePages {
+            let pagesToFetch = Array(currentPage..<(currentPage + batchSize))
+
+            let batchResults = try await withThrowingTaskGroup(of: PageResult<POSProduct>.self) { group in
+                for pageNumber in pagesToFetch {
+                    group.addTask {
+                        let result = try await syncRemote.loadProducts(siteID: siteID, pageNumber: pageNumber)
+                        return PageResult(pageNumber: pageNumber, items: result)
+                    }
+                }
+
+                var results: [PageResult<POSProduct>] = []
+                for try await result in group {
+                    results.append(result)
+                }
+                return results.sorted(by: { $0.pageNumber < $1.pageNumber })
+            }
+
+            // Processes results in order and checks if there are more pages.
+            let newProducts = batchResults.flatMap { $0.items.items }
+            allProducts.append(contentsOf: newProducts)
+
+            let highestPageResult = batchResults.last?.items
+            hasMorePages = (highestPageResult?.hasMorePages ?? false) && !newProducts.isEmpty
+            currentPage += batchSize
+
+            DDLogInfo("📥 Loaded batch: \(batchResults.count) pages, total products: \(allProducts.count), hasMorePages: \(hasMorePages)")
+        }
+
+        DDLogInfo("✅ Products sync complete: \(allProducts.count) products loaded")
+        return allProducts
+    }
+
+    func loadAllProductVariations(for siteID: Int64, syncRemote: POSCatalogSyncRemoteProtocol) async throws -> [POSProductVariation] {
+        DDLogInfo("🔄 Starting variations sync for site ID: \(siteID)")
+
+        var allVariations: [POSProductVariation] = []
+        var currentPage = 1
+        var hasMorePages = true
+
+        while hasMorePages {
+            let pagesToFetch = Array(currentPage..<(currentPage + batchSize))
+
+            let batchResults = try await withThrowingTaskGroup(of: PageResult<POSProductVariation>.self) { group in
+                for pageNumber in pagesToFetch {
+                    group.addTask {
+                        let result = try await syncRemote.loadProductVariations(siteID: siteID, pageNumber: pageNumber)
+                        return PageResult(pageNumber: pageNumber, items: result)
+                    }
+                }
+
+                var results: [PageResult<POSProductVariation>] = []
+                for try await result in group {
+                    results.append(result)
+                }
+                return results.sorted(by: { $0.pageNumber < $1.pageNumber })
+            }
+
+            // Processes results in order and checks if there are more pages.
+            let newVariations = batchResults.flatMap { $0.items.items }
+            allVariations.append(contentsOf: newVariations)
+
+            let highestPageResult = batchResults.last?.items
+            hasMorePages = (highestPageResult?.hasMorePages ?? false) && !newVariations.isEmpty
+            currentPage += batchSize
+
+            DDLogInfo("📥 Loaded batch: \(batchResults.count) pages, total variations: \(allVariations.count), hasMorePages: \(hasMorePages)")
+        }
+
+        DDLogInfo("✅ Variations sync complete: \(allVariations.count) variations loaded")
+        return allVariations
+    }
+}
+
+private struct PageResult<T> {
+    let pageNumber: Int
+    let items: PagedItems<T>
+}
diff --git a/Modules/Tests/NetworkingTests/Network/AlamofireNetworkTests.swift b/Modules/Tests/NetworkingTests/Network/AlamofireNetworkTests.swift
@@ -167,6 +167,50 @@ final class AlamofireNetworkTests: XCTestCase {
         // Then
         XCTAssertTrue(result.isSuccess)
     }
+
+    // MARK: - Session Initialization Tests
+
+    func test_concurrent_requests_do_not_fail_with_sessionDeinitialized_error_when_ensuresSessionManagerIsInitialized_is_true() async throws {
+        // Given
+        let request = JetpackRequest(wooApiVersion: .mark1, method: .get, siteID: -1, path: "test")
+        let network = AlamofireNetwork(credentials: nil, ensuresSessionManagerIsInitialized: true)
+
+        // When
+        async let request1 = network.responseDataAndHeaders(for: request)
+        async let request2 = network.responseDataAndHeaders(for: request)
+        async let request3 = network.responseDataAndHeaders(for: request)
+
+        do {
+            _ = try await [request1, request2, request3]
+            XCTFail("Requests should fail")
+        } catch Alamofire.AFError.sessionDeinitialized {
+            XCTFail("Requests should not fail with sessionDeinitialized error")
+        } catch {
+            // Then
+            XCTAssertTrue(true)
+        }
+    }
+
+    func test_concurrent_requests_fail_with_sessionDeinitialized_error_when_ensuresSessionManagerIsInitialized_is_false() async throws {
+        // Given
+        let request = JetpackRequest(wooApiVersion: .mark1, method: .get, siteID: 1, path: "test")
+        let network = AlamofireNetwork(credentials: nil, ensuresSessionManagerIsInitialized: false)
+
+        // When
+        async let request1 = network.responseDataAndHeaders(for: request)
+        async let request2 = network.responseDataAndHeaders(for: request)
+        async let request3 = network.responseDataAndHeaders(for: request)
+
+        do {
+            _ = try await [request1, request2, request3]
+            XCTFail("Requests should fail with sessionDeinitialized error")
+        } catch Alamofire.AFError.sessionDeinitialized {
+            // Then
+            XCTAssertTrue(true)
+        } catch {
+            XCTFail("Requests should fail with sessionDeinitialized error")
+        }
+    }
 }
 
 private extension AlamofireNetworkTests {
diff --git a/Modules/Tests/YosemiteTests/Tools/POS/POSCatalogFullSyncServiceTests.swift b/Modules/Tests/YosemiteTests/Tools/POS/POSCatalogFullSyncServiceTests.swift
