Observe catalog sync state using Observation model

Author: staskus

Modules/Sources/PointOfSale/Models/PointOfSaleAggregateModel.swift

@@ -16,6 +16,7 @@ import protocol Yosemite.PointOfSaleBarcodeScanServiceProtocol
 import enum Yosemite.PointOfSaleBarcodeScanError
 import protocol Yosemite.POSCatalogSyncCoordinatorProtocol
 import enum Yosemite.POSCatalogSyncState
+import class Yosemite.POSCatalogSyncCoordinator
 
 protocol PointOfSaleAggregateModelProtocol {
     var cart: Cart { get }
@@ -63,6 +64,7 @@ protocol PointOfSaleAggregateModelProtocol {
     private let soundPlayer: PointOfSaleSoundPlayerProtocol
 
     private var cancellables: Set<AnyCancellable> = []
+    private var catalogSyncStateTask: Task<Void, Never>?
 
     // Private storage of the concrete coordinator
     private let _viewStateCoordinator = PointOfSaleViewStateCoordinator()
@@ -117,7 +119,7 @@ protocol PointOfSaleAggregateModelProtocol {
         setupReaderReconnectionObservation()
         setupPaymentSuccessObservation()
         performIncrementalSync()
-        publishCatalogSyncState()
+        setupCatalogSyncStateObservation()
     }
 }
 
@@ -601,6 +603,7 @@ extension PointOfSaleAggregateModel {
         // cancelling them explicitly helps reduce the risk of user-visible bugs while we work on the memory leaks.
         resetCardReaderObservation()
         cancellables.forEach { $0.cancel() }
+        catalogSyncStateTask?.cancel()
     }
 }
 
@@ -630,17 +633,26 @@ private extension PointOfSaleAggregateModel {
 // MARK: - Catalog Sync State Observation
 
 private extension PointOfSaleAggregateModel {
-    private func publishCatalogSyncState() {
+    private func setupCatalogSyncStateObservation() {
         guard let coordinator = catalogSyncCoordinator else { return }
 
-        Task { @MainActor [weak self] in
-            guard let self else { return }
-
+        Task { @MainActor in
             catalogSyncState = await coordinator.lastFullSyncState(for: siteID)
+        }
+
+        observeCatalogSyncState()
+    }
 
-            for await state in coordinator.fullSyncStateStream {
+    @Sendable private func observeCatalogSyncState() {
+        withObservationTracking { [weak self] in
+            guard let self else { return }
+
+            if let state =  catalogSyncCoordinator?.fullSyncStateModel.state[siteID] {
                 catalogSyncState = state
             }
+        } onChange: { [weak self] in
+            guard let self else { return }
+            DispatchQueue.main.async(execute: observeCatalogSyncState)
         }
     }
 }

Modules/Sources/PointOfSale/Utils/PreviewHelpers.swift

@@ -30,6 +30,7 @@ import class Yosemite.POSOrderListService
 import class Yosemite.POSOrderListFetchStrategyFactory
 import protocol Yosemite.POSCatalogSyncCoordinatorProtocol
 import enum Yosemite.POSCatalogSyncState
+import class Yosemite.POSCatalogSyncStateModel
 import protocol Yosemite.POSCatalogSettingsServiceProtocol
 import struct Yosemite.POSCatalogInfo
 import struct Yosemite.Site
@@ -629,13 +630,10 @@ final class POSPreviewCatalogSyncCoordinator: POSCatalogSyncCoordinatorProtocol
         try await Task.sleep(nanoseconds: 1_000_000_000)
     }
 
-    let fullSyncStateStream: AsyncStream<POSCatalogSyncState> = {
-        let (stream, _) = AsyncStream<POSCatalogSyncState>.makeStream()
-        return stream
-    }()
+    let fullSyncStateModel = POSCatalogSyncStateModel()
 
     func lastFullSyncState(for siteID: Int64) async -> POSCatalogSyncState {
-        return .syncCompleted(siteID: siteID)
+        return fullSyncStateModel.state[siteID] ?? .syncCompleted(siteID: siteID)
     }
 }
 

Modules/Sources/Yosemite/Tools/POS/POSCatalogSyncCoordinator.swift

@@ -29,7 +29,7 @@ public protocol POSCatalogSyncCoordinatorProtocol {
     func performSmartSync(for siteID: Int64, fullSyncMaxAge: TimeInterval, incrementalSyncMaxAge: TimeInterval) async throws
 
     /// Stream that emits full sync state updates
-    var fullSyncStateStream: AsyncStream<POSCatalogSyncState> { get }
+    var fullSyncStateModel: POSCatalogSyncStateModel { get }
 
     /// Returns the last known full sync state for a site
     /// If no state is cached, determines state from lastSyncDate
@@ -70,12 +70,8 @@ public actor POSCatalogSyncCoordinator: POSCatalogSyncCoordinatorProtocol {
     /// Tracks ongoing incremental syncs by site ID to prevent duplicates
     private var ongoingIncrementalSyncs: Set<Int64> = []
 
-    /// Stream for full sync state updates
-    public nonisolated let fullSyncStateStream: AsyncStream<POSCatalogSyncState>
-    /// Continuation for emitting state updates
-    private let fullSyncStateStreamContinuation: AsyncStream<POSCatalogSyncState>.Continuation
-    /// Cache of last known full sync state for each site
-    private var fullSyncStateCache: [Int64: POSCatalogSyncState] = [:]
+    /// Observable model for full sync state updates
+    public nonisolated let fullSyncStateModel: POSCatalogSyncStateModel = .init()
 
     public init(fullSyncService: POSCatalogFullSyncServiceProtocol,
                 incrementalSyncService: POSCatalogIncrementalSyncServiceProtocol,
@@ -87,10 +83,6 @@ public actor POSCatalogSyncCoordinator: POSCatalogSyncCoordinatorProtocol {
         self.grdbManager = grdbManager
         self.catalogSizeLimit = catalogSizeLimit ?? Constants.defaultSizeLimitForPOSCatalog
         self.catalogSizeChecker = catalogSizeChecker
-
-        let (stream, continuation) = AsyncStream<POSCatalogSyncState>.makeStream()
-        self.fullSyncStateStream = stream
-        self.fullSyncStateStreamContinuation = continuation
     }
 
     public func performFullSyncIfApplicable(for siteID: Int64, maxAge: TimeInterval) async throws {
@@ -102,7 +94,7 @@ public actor POSCatalogSyncCoordinator: POSCatalogSyncCoordinatorProtocol {
             return
         }
 
-        if case .syncStarted = fullSyncStateCache[siteID] {
+        if case .syncStarted = fullSyncStateModel.state[siteID] {
             DDLogInfo("⚠️ POSCatalogSyncCoordinator: Sync already in progress for site \(siteID)")
             throw POSCatalogSyncError.syncAlreadyInProgress(siteID: siteID)
         }
@@ -309,7 +301,7 @@ public actor POSCatalogSyncCoordinator: POSCatalogSyncCoordinatorProtocol {
     }
 
     public func lastFullSyncState(for siteID: Int64) async -> POSCatalogSyncState {
-        if let cached = fullSyncStateCache[siteID] {
+        if let cached = fullSyncStateModel.state[siteID] {
             return cached
         }
 
@@ -330,11 +322,17 @@ private extension POSCatalogSyncCoordinator {
             id
         }
 
-        fullSyncStateCache[siteID] = state
-        fullSyncStateStreamContinuation.yield(state)
+        fullSyncStateModel.state[siteID] = state
     }
 }
 
+@Observable
+public class POSCatalogSyncStateModel {
+    public var state: [Int64: POSCatalogSyncState] = [:]
+
+    public init() {}
+}
+
 public enum POSCatalogSyncState: Equatable {
     case syncStarted(siteID: Int64, isInitialSync: Bool)
     case syncCompleted(siteID: Int64)

Modules/Tests/PointOfSaleTests/Mocks/MockPOSCatalogSyncCoordinator.swift

@@ -63,12 +63,9 @@ final class MockPOSCatalogSyncCoordinator: POSCatalogSyncCoordinatorProtocol {
         }
     }
 
-    let fullSyncStateStream: AsyncStream<POSCatalogSyncState> = {
-        let (stream, _) = AsyncStream<POSCatalogSyncState>.makeStream()
-        return stream
-    }()
+    let fullSyncStateModel = POSCatalogSyncStateModel()
 
     func lastFullSyncState(for siteID: Int64) async -> POSCatalogSyncState {
-        return .syncNeverDone(siteID: siteID)
+        return fullSyncStateModel.state[siteID] ?? .syncNeverDone(siteID: siteID)
     }
 }

Modules/Tests/YosemiteTests/Tools/POS/POSCatalogSyncCoordinatorTests.swift

@@ -343,8 +343,8 @@ struct POSCatalogSyncCoordinatorTests {
 
     @Test func performIncrementalSyncIfApplicable_skips_sync_when_incremental_sync_is_within_max_age() async throws {
         // Given
-        let maxAge: TimeInterval = 2
-        let incrementalSyncDate = Date().addingTimeInterval(-(maxAge - 0.2)) // Just within max age
+        let maxAge: TimeInterval = 60 // 60 seconds
+        let incrementalSyncDate = Date().addingTimeInterval(-30) // 30 seconds ago (within maxAge)
         let fullSyncDate = Date().addingTimeInterval(-7200)
         try createSiteInDatabase(siteID: sampleSiteID, lastFullSyncDate: fullSyncDate, lastIncrementalSyncDate: incrementalSyncDate)
 
@@ -358,7 +358,7 @@ struct POSCatalogSyncCoordinatorTests {
         // When
         try await sut.performIncrementalSyncIfApplicable(for: sampleSiteID, maxAge: maxAge)
 
-        // Then
+        // Then - should not sync because 30 seconds < 60 second maxAge
         #expect(mockIncrementalSyncService.startIncrementalSyncCallCount == 0)
     }
 
@@ -716,32 +716,16 @@ struct POSCatalogSyncCoordinatorTests {
         #expect(state == .syncCompleted(siteID: sampleSiteID))
     }
 
-    @Test func fullSyncStateStream_emits_events_during_sync() async throws {
+    @Test func fullSyncStateModel_emits_events_during_sync() async throws {
         // Given
         let expectedCatalog = POSCatalog(products: [], variations: [], syncDate: .now)
         mockSyncService.startFullSyncResult = .success(expectedCatalog)
 
         // When - start sync and stream collection concurrently
-        async let syncResult: Void = sut.performFullSync(for: sampleSiteID)
-
-        var receivedEvents: [POSCatalogSyncState] = []
-        let collectTask = Task {
-            for await state in sut.fullSyncStateStream {
-                receivedEvents.append(state)
-                // Stop after getting completion
-                if case .syncCompleted(let id) = state, id == self.sampleSiteID {
-                    break
-                }
-            }
-        }
-
-        // Wait for sync and stream collection
-        try await syncResult
-        _ = await collectTask.value
+        try await sut.performFullSync(for: sampleSiteID)
 
         // Then - should emit syncStarted and syncCompleted with correct siteID
-        #expect(receivedEvents.contains(where: { if case .syncStarted(let id, _) = $0 { return id == self.sampleSiteID } else { return false } }))
-        #expect(receivedEvents.contains { $0 == .syncCompleted(siteID: self.sampleSiteID) })
+        #expect(sut.fullSyncStateModel.state[sampleSiteID] == .syncCompleted(siteID: sampleSiteID))
     }
 
     // MARK: - Helper Methods

WooCommerce/WooCommerceTests/Tools/ForegroundPOSCatalogSyncDispatcherTests.swift

@@ -292,12 +292,9 @@ private final class MockPOSCatalogSyncCoordinator: POSCatalogSyncCoordinatorProt
         // Not used
     }
 
-    let fullSyncStateStream: AsyncStream<POSCatalogSyncState> = {
-        let (stream, _) = AsyncStream<POSCatalogSyncState>.makeStream()
-        return stream
-    }()
+    let fullSyncStateModel = POSCatalogSyncStateModel()
 
     func lastFullSyncState(for siteID: Int64) async -> POSCatalogSyncState {
-        return .syncNeverDone(siteID: siteID)
+        return fullSyncStateModel.state[siteID] ?? .syncNeverDone(siteID: siteID)
     }
 }