[FME-4230] Events - Updated Segments

Split/Events/SplitEventsManager.swift

@@ -51,6 +51,7 @@ class DefaultSplitEventsManager: SplitEventsManager {
         }
     }
 
+    // MARK: Notifiers
     func notifyInternalEvent(_ event: SplitInternalEvent, metadata: EventMetadata? = nil) {
         let event = SplitInternalEventWithMetadata(event, metadata: metadata)
 
@@ -64,7 +65,8 @@ class DefaultSplitEventsManager: SplitEventsManager {
     func notifyInternalEvent(_ event: SplitInternalEvent) {
         notifyInternalEvent(event, metadata: nil)
     }
-
+
+    // MARK: Registers
     func register(event: SplitEventWithMetadata, task: SplitEventActionTask) {
         let eventName = event.type.toString()
         processQueue.async { [weak self] in
@@ -83,6 +85,7 @@ class DefaultSplitEventsManager: SplitEventsManager {
         register(event: SplitEventWithMetadata(type: event, metadata: nil), task: task)
     }
 
+    // MARK: Flow
     func start() {
         dataAccessQueue.sync {
             if self.isStarted {
@@ -158,31 +161,31 @@ class DefaultSplitEventsManager: SplitEventsManager {
             }
             self.triggered.append(event.type)
             switch event.type {
-            case .splitsUpdated, .mySegmentsUpdated, .myLargeSegmentsUpdated:
-                if isTriggered(external: .sdkReady) {
-                    trigger(event: .sdkUpdated, metadata: event.metadata)
-                    continue
-                }
-                self.triggerSdkReadyIfNeeded()
-
-            case .mySegmentsLoadedFromCache, .myLargeSegmentsLoadedFromCache,
-                    .splitsLoadedFromCache, .attributesLoadedFromCache:
-                Logger.v("Event \(event) triggered")
-                if isTriggered(internal: .splitsLoadedFromCache),
-                   isTriggered(internal: .mySegmentsLoadedFromCache),
-                   isTriggered(internal: .myLargeSegmentsLoadedFromCache),
-                   isTriggered(internal: .attributesLoadedFromCache) {
-                    trigger(event: .sdkReadyFromCache)
-                }
-            case .splitKilledNotification:
-                if isTriggered(external: .sdkReady) {
-                    trigger(event: .sdkUpdated, metadata: event.metadata)
-                    continue
-                }
-            case .sdkReadyTimeoutReached:
-                if !isTriggered(external: .sdkReady) {
-                    trigger(event: .sdkReadyTimedOut)
-                }
+                case .splitsUpdated, .mySegmentsUpdated, .myLargeSegmentsUpdated:
+                    if isTriggered(external: .sdkReady) {
+                        trigger(event: .sdkUpdated, metadata: event.metadata)
+                        continue
+                    }
+                    self.triggerSdkReadyIfNeeded()
+
+                case .mySegmentsLoadedFromCache, .myLargeSegmentsLoadedFromCache,
+                        .splitsLoadedFromCache, .attributesLoadedFromCache:
+                    Logger.v("Event \(event) triggered")
+                    if isTriggered(internal: .splitsLoadedFromCache),
+                       isTriggered(internal: .mySegmentsLoadedFromCache),
+                       isTriggered(internal: .myLargeSegmentsLoadedFromCache),
+                       isTriggered(internal: .attributesLoadedFromCache) {
+                        trigger(event: .sdkReadyFromCache)
+                    }
+                case .splitKilledNotification:
+                    if isTriggered(external: .sdkReady) {
+                        trigger(event: .sdkUpdated, metadata: event.metadata)
+                        continue
+                    }
+                case .sdkReadyTimeoutReached:
+                    if !isTriggered(external: .sdkReady) {
+                        trigger(event: .sdkReadyTimedOut)
+                    }
             }
         }
     }

Split/FetcherEngine/Refresh/ChangesChecker.swift

@@ -22,6 +22,7 @@ protocol MySegmentsChangesChecker {
     func mySegmentsHaveChanged(old: SegmentChange, new: SegmentChange) -> Bool
     func mySegmentsHaveChanged(oldSegments: [Segment], newSegments: [Segment]) -> Bool
     func mySegmentsHaveChanged(oldSegments: [String], newSegments: [String]) -> Bool
+    func getSegmentsDiff(oldSegments: [Segment], newSegments: [Segment]) -> [String]
 }
 
 struct DefaultMySegmentsChangesChecker: MySegmentsChangesChecker {
@@ -42,5 +43,8 @@ struct DefaultMySegmentsChangesChecker: MySegmentsChangesChecker {
         return !(oldSegments.count == newSegments.count &&
                  oldSegments.sorted() == newSegments.sorted())
     }
-
+
+    func getSegmentsDiff(oldSegments: [Segment], newSegments: [Segment]) -> [String] {
+        oldSegments.filter { !Set(newSegments.map { $0.name }).contains($0.name) }.map { $0.name }
+    }
 }

Split/FetcherEngine/Refresh/PeriodicSyncWorker.swift

@@ -130,7 +130,7 @@ class BasePeriodicSyncWorker: PeriodicSyncWorker {
         Logger.i("Fetch from remote not implemented")
     }
 
-    func notifyUpdate(_ event: SplitInternalEvent, _ metadata: EventMetadata? = nil) {
+    func notifyUpdate(_ event: SplitInternalEvent, metadata: EventMetadata? = nil) {
         eventsManager.notifyInternalEvent(event, metadata: metadata)
     }
 }
@@ -184,7 +184,7 @@ class PeriodicSplitsSyncWorker: BasePeriodicSyncWorker {
 
         if result.success, result.featureFlagsUpdated.count > 0 {
             let metadata = EventMetadata(type: .FLAGS_UPDATED, data: result.featureFlagsUpdated)
-            notifyUpdate(.splitsUpdated, metadata)
+            notifyUpdate(.splitsUpdated, metadata: metadata)
         }
     }
 }
@@ -223,9 +223,14 @@ class PeriodicMySegmentsSyncWorker: BasePeriodicSyncWorker {
                                              mlsTill: myLargeSegmentsStorage.changeNumber,
                                              headers: nil)
             if result.success {
-                if  result.msUpdated || result.mlsUpdated {
+                if !result.msUpdated.isEmpty || !result.mlsUpdated.isEmpty {
                     // For now is not necessary specify which entity was updated
-                    notifyUpdate(.mySegmentsUpdated)
+                    if !result.msUpdated.isEmpty {
+                        notifyUpdate(.mySegmentsUpdated, metadata: EventMetadata(type: .SEGMENTS_UPDATED, data: result.msUpdated))
+                    }
+                    if !result.mlsUpdated.isEmpty {
+                        notifyUpdate(.myLargeSegmentsUpdated, metadata: EventMetadata(type: .LARGE_SEGMENTS_UPDATED, data: result.msUpdated))
+                    }
                 }
             }
         } catch {

Split/FetcherEngine/Refresh/RetryableSegmentsSyncWorker.swift

@@ -46,11 +46,16 @@ class RetryableMySegmentsSyncWorker: BaseRetryableSyncWorker {
             if result.success {
                 if !isSdkReadyTriggered() {
                     // Notifying both to trigger SDK Ready
-                    notifyUpdate(.mySegmentsUpdated)
-                    notifyUpdate(.myLargeSegmentsUpdated)
-                } else if  result.msUpdated || result.mlsUpdated {
+                    notifyUpdate(.mySegmentsUpdated, metadata: EventMetadata(type: .SEGMENTS_UPDATED, data: result.msUpdated))
+                    notifyUpdate(.myLargeSegmentsUpdated, metadata: EventMetadata(type: .LARGE_SEGMENTS_UPDATED, data: result.mlsUpdated))
+                } else if !result.msUpdated.isEmpty || !result.mlsUpdated.isEmpty {
                     // For now is not necessary specify which entity was updated
-                    notifyUpdate(.mySegmentsUpdated)
+                    if !result.msUpdated.isEmpty {
+                        notifyUpdate(.mySegmentsUpdated, metadata: EventMetadata(type: .SEGMENTS_UPDATED, data: result.msUpdated))
+                    }
+                    if !result.mlsUpdated.isEmpty {
+                        notifyUpdate(.myLargeSegmentsUpdated, metadata: EventMetadata(type: .LARGE_SEGMENTS_UPDATED, data: result.mlsUpdated))
+                    }
                 }
                 return true
             }
@@ -71,8 +76,8 @@ struct SegmentsSyncResult {
     let success: Bool
     let msChangeNumber: Int64
     let mlsChangeNumber: Int64
-    let msUpdated: Bool
-    let mlsUpdated: Bool
+    let msUpdated: [String]
+    let mlsUpdated: [String]
 }
 
 protocol SegmentsSyncHelper {
@@ -86,8 +91,8 @@ class DefaultSegmentsSyncHelper: SegmentsSyncHelper {
     struct FetchResult {
         let msTill: Int64
         let mlsTill: Int64
-        let msUpdated: Bool
-        let mlsUdated: Bool
+        let msUpdated: [String]
+        let mlsUpdated: [String]
     }
 
     private let segmentsFetcher: HttpMySegmentsFetcher
@@ -165,7 +170,7 @@ class DefaultSegmentsSyncHelper: SegmentsSyncHelper {
                                           msChangeNumber: result.msTill,
                                           mlsChangeNumber: result.mlsTill,
                                           msUpdated: result.msUpdated,
-                                          mlsUpdated: result.mlsUdated)
+                                          mlsUpdated: result.mlsUpdated)
             }
             attemptCount+=1
             if attemptCount < maxAttempts {
@@ -175,8 +180,8 @@ class DefaultSegmentsSyncHelper: SegmentsSyncHelper {
         return SegmentsSyncResult(success: false,
                                   msChangeNumber: -1,
                                   mlsChangeNumber: -1,
-                                  msUpdated: false,
-                                  mlsUpdated: false)
+                                  msUpdated: [],
+                                  mlsUpdated: [])
     }
 
     private func fetchUntil(till: Int64?,
@@ -210,10 +215,13 @@ class DefaultSegmentsSyncHelper: SegmentsSyncHelper {
                 Logger.d("Checking my large segments update")
                 checkAndUpdate(isChanged: mlsChanged, change: myLargeSegmentsChange, storage: myLargeSegmentsStorage)
 
+                let segmentsDiff = changeChecker.getSegmentsDiff(oldSegments: oldChange.segments, newSegments: mySegmentsChange.segments)
+                let largeSegmentsDiff = changeChecker.getSegmentsDiff(oldSegments: oldLargeChange.segments, newSegments: myLargeSegmentsChange.segments)
+
                 return FetchResult(msTill: mySegmentsChange.unwrappedChangeNumber,
                                    mlsTill: myLargeSegmentsChange.unwrappedChangeNumber,
-                                   msUpdated: msChanged,
-                                   mlsUdated: mlsChanged)
+                                   msUpdated: segmentsDiff,
+                                   mlsUpdated: largeSegmentsDiff)
             }
             prevChange = change
         }

Split/Localhost/LocalhostSynchronizer.swift

@@ -74,7 +74,9 @@ class LocalhostSynchronizer: FeatureFlagsSynchronizer {
             // Update will remove all records before insert new ones
             _ = self.featureFlagsStorage.update(splitChange: change)
 
-            self.eventsManager.notifyInternalEvent(.splitsUpdated, metadata: EventMetadata(type: .FLAGS_UPDATED, data: values.map { $0.name ?? "" } ))
+            // Notify event
+            let metadata = EventMetadata(type: .FLAGS_UPDATED, data: values.map { $0.name ?? "" } )
+            self.eventsManager.notifyInternalEvent(.splitsUpdated, metadata: metadata)
         }
     }
 }

Split/Network/Streaming/PushNotificationManager.swift

@@ -134,14 +134,7 @@ class DefaultPushNotificationManager: PushNotificationManager {
         }
         Logger.d("Streaming authentication success")
 
-        var connectionDelay: Int64 = 0
-
-        #if DEBUG
-            connectionDelay = 1
-        #else
-            connectionDelay = result.sseConnectionDelay
-        #endif
-
+        let connectionDelay = result.sseConnectionDelay
         self.broadcasterChannel.push(event: .pushDelayReceived(delaySeconds: connectionDelay))
         let lastId = lastConnId.value
         if connectionDelay > 0 {

Split/Network/Streaming/SyncSegmentsUpdateWorker.swift

@@ -110,7 +110,7 @@ class SegmentsUpdateWorker: UpdateWorker<MembershipsUpdateNotification> {
         if segments.count > newSegments.count {
             mySegmentsStorage.set(SegmentChange(segments: newSegments.asArray()),
                                   forKey: key)
-            synchronizer.notifyUpdate(forKey: key)
+            synchronizer.notifyUpdate(forKey: key, EventMetadata(type: .SEGMENTS_UPDATED, data: newSegments.asArray() ))
             telemetryProducer?.recordUpdatesFromSse(type: resource)
         }
     }
@@ -128,7 +128,7 @@ class SegmentsUpdateWorker: UpdateWorker<MembershipsUpdateNotification> {
                 if oldSegments.count < newSegments.count {
                     mySegmentsStorage.set(SegmentChange(segments: newSegments.asArray()),
                                           forKey: userKey)
-                    synchronizer.notifyUpdate(forKey: userKey)
+                    synchronizer.notifyUpdate(forKey: userKey, EventMetadata(type: .SEGMENTS_UPDATED, data: newSegments.asArray() ))
                     telemetryProducer?.recordUpdatesFromSse(type: .mySegments)
                 }
                 return
@@ -171,7 +171,7 @@ class SegmentsUpdateWorker: UpdateWorker<MembershipsUpdateNotification> {
 
 protocol SegmentsSynchronizerWrapper {
     func fetch(byKey: String, changeNumbers: SegmentsChangeNumber, delay: Int64)
-    func notifyUpdate(forKey: String)
+    func notifyUpdate(forKey: String, _ metadata: EventMetadata?)
 }
 
 class MySegmentsSynchronizerWrapper: SegmentsSynchronizerWrapper {
@@ -185,12 +185,13 @@ class MySegmentsSynchronizerWrapper: SegmentsSynchronizerWrapper {
         synchronizer.forceMySegmentsSync(forKey: key, changeNumbers: changeNumbers, delay: delay)
     }
 
-    func notifyUpdate(forKey key: String) {
-        synchronizer.notifySegmentsUpdated(forKey: key)
+    func notifyUpdate(forKey key: String, _ metadata: EventMetadata? = nil) {
+        synchronizer.notifySegmentsUpdated(forKey: key, metadata: metadata)
     }
 }
 
 class MyLargeSegmentsSynchronizerWrapper: SegmentsSynchronizerWrapper {
+
     private let synchronizer: Synchronizer
 
     init(synchronizer: Synchronizer) {
@@ -201,8 +202,8 @@ class MyLargeSegmentsSynchronizerWrapper: SegmentsSynchronizerWrapper {
         synchronizer.forceMySegmentsSync(forKey: key, changeNumbers: changeNumbers, delay: delay)
     }
 
-    func notifyUpdate(forKey key: String) {
-        synchronizer.notifyLargeSegmentsUpdated(forKey: key)
+    func notifyUpdate(forKey key: String, _ metadata: EventMetadata? = nil) {
+        synchronizer.notifyLargeSegmentsUpdated(forKey: key, metadata: metadata)
     }
 }
 

Split/Network/Streaming/SyncUpdateWorker.swift

@@ -169,14 +169,15 @@ class SplitsUpdateWorker: UpdateWorker<TargetingRuleUpdateNotification> {
 
             Logger.v("RBS update received: \(change)")
 
-            let processedChange = ruleBasedSegmentsChangeProcessor.process(change)
+            let processedSegments = ruleBasedSegmentsChangeProcessor.process(change)
 
-            if ruleBasedSegmentsStorage.update(toAdd: processedChange.toAdd,
-                                               toRemove: processedChange.toRemove,
-                                               changeNumber: processedChange.changeNumber) {
-                synchronizer.notifyFeatureFlagsUpdated(flags: []) //TODO: RBS Update
+            if ruleBasedSegmentsStorage.update(toAdd: processedSegments.toAdd,
+                                                  toRemove: processedSegments.toRemove,
+                                                  changeNumber: processedSegments.changeNumber) {
+                var updatedSegments = (processedSegments.activeSegments + processedSegments.archivedSegments).compactMap(\.name)
+                synchronizer.notifyFeatureFlagsUpdated(flags: []) //TODO: Make new notify segments updated (new notification method?)
             }
-
+            
             telemetryProducer?.recordUpdatesFromSse(type: .splits)
             return true
         } catch {

Split/Network/Sync/ByKeyFacade.swift

@@ -19,8 +19,8 @@ protocol ByKeyRegistry {
 protocol ByKeySynchronizer {
     func loadMySegmentsFromCache(forKey: String)
     func loadAttributesFromCache(forKey: String)
-    func notifyMySegmentsUpdated(forKey: String)
-    func notifyMyLargeSegmentsUpdated(forKey: String)
+    func notifyMySegmentsUpdated(forKey: String, metadata: EventMetadata?)
+    func notifyMyLargeSegmentsUpdated(forKey: String, metadata: EventMetadata?)
     func startSync(forKey key: Key)
     func startPeriodicSync()
     func stopPeriodicSync()
@@ -120,15 +120,15 @@ class DefaultByKeyFacade: ByKeyFacade {
         byKeyComponents.value(forKey: key)?.mySegmentsSynchronizer.synchronizeMySegments()
     }
 
-    func notifyMySegmentsUpdated(forKey key: String) {
+    func notifyMySegmentsUpdated(forKey key: String, metadata: EventMetadata?) {
         doInAll(forMatchingKey: key) { group in
-            group.eventsManager.notifyInternalEvent(.mySegmentsUpdated)
+            group.eventsManager.notifyInternalEvent(.mySegmentsUpdated, metadata: metadata)
         }
     }
 
-    func notifyMyLargeSegmentsUpdated(forKey key: String) {
+    func notifyMyLargeSegmentsUpdated(forKey key: String, metadata: EventMetadata? = nil) {
         doInAll(forMatchingKey: key) { group in
-            group.eventsManager.notifyInternalEvent(.myLargeSegmentsUpdated)
+            group.eventsManager.notifyInternalEvent(.myLargeSegmentsUpdated, metadata: metadata)
         }
     }
 

Split/Network/Sync/Synchronizer.swift

@@ -31,8 +31,8 @@ protocol Synchronizer: ImpressionLogger {
     func stopRecordingTelemetry()
     func pushEvent(event: EventDTO)
     func notifyFeatureFlagsUpdated(flags: [String])
-    func notifySegmentsUpdated(forKey key: String)
-    func notifyLargeSegmentsUpdated(forKey key: String)
+    func notifySegmentsUpdated(forKey key: String, metadata: EventMetadata?)
+    func notifyLargeSegmentsUpdated(forKey key: String, metadata: EventMetadata?)
     func notifySplitKilled(flag: String)
     func pause()
     func resume()
@@ -211,12 +211,12 @@ class DefaultSynchronizer: Synchronizer {
         featureFlagsSynchronizer.notifyUpdated(flagsList: flags)
     }
 
-    func notifySegmentsUpdated(forKey key: String) {
-        byKeySynchronizer.notifyMySegmentsUpdated(forKey: key)
+    func notifySegmentsUpdated(forKey key: String, metadata: EventMetadata? = nil) {
+        byKeySynchronizer.notifyMySegmentsUpdated(forKey: key, metadata: metadata)
     }
 
-    func notifyLargeSegmentsUpdated(forKey key: String) {
-        byKeySynchronizer.notifyMyLargeSegmentsUpdated(forKey: key)
+    func notifyLargeSegmentsUpdated(forKey key: String, metadata: EventMetadata? = nil) {
+        byKeySynchronizer.notifyMyLargeSegmentsUpdated(forKey: key, metadata: metadata)
     }
 
     func notifySplitKilled(flag: String) {

SplitTests/Fake/Service/ByKeyFacadeMock.swift

@@ -123,12 +123,16 @@ class ByKeyFacadeMock: ByKeyFacade {
     }
 
     var notifyMySegmentsUpdatedCalled = false
-    func notifyMySegmentsUpdated(forKey key: String) {
+    var updatedSegmentsMetadataForKey = [String : EventMetadata?]()
+    func notifyMySegmentsUpdated(forKey key: String, metadata: EventMetadata? = nil) {
+        updatedSegmentsMetadataForKey[key] = metadata
         notifyMySegmentsUpdatedCalled = true
     }
 
     var notifyMyLargeSegmentsUpdatedCalled = false
-    func notifyMyLargeSegmentsUpdated(forKey key: String) {
+    var updatedLargeSegmentsMetadataForKey = [String : EventMetadata?]()
+    func notifyMyLargeSegmentsUpdated(forKey key: String, metadata: EventMetadata? = nil) {
+        updatedLargeSegmentsMetadataForKey[key] = metadata
         notifyMyLargeSegmentsUpdatedCalled = true
     }