PIR: support remove defunct brokers (#6568)

Task/Issue URL:
https://app.asana.com/1/137249556945/project/488551667048375/task/1211032016934484?focus=true

### Description
See attached task description

### Steps to test this PR
https://app.asana.com/1/137249556945/task/1211035661205077?focus=true

Author: karlenDimla

pir/pir-impl/schemas/com.duckduckgo.pir.internal.store.PirDatabase/1.json renamed to pir/pir-impl/schemas/com.duckduckgo.pir.impl.store.PirDatabase/1.json

@@ -92,26 +92,23 @@ class RealBrokerJsonUpdater @Inject constructor(
     }
 
     private suspend fun checkUpdatesFromMainConfig(pirMainConfig: PirMainConfig) {
-        val currentBrokerJson = pirRepository.getAllLocalBrokerJsons()
-        val newActiveBrokers = pirMainConfig.activeBrokers
-        val newBrokerJson = pirMainConfig.jsonEtags.current.map {
+        val existingEtags = pirRepository.getAllLocalBrokerJsons()
+        val jsonEtagsFromConfig = pirMainConfig.jsonEtags.current.map {
             BrokerJson(
                 fileName = it.key,
                 etag = it.value,
             )
-        }.associateWith { newActiveBrokers.contains(it.fileName) }
+        }
 
-        val shouldDownloadJson = currentBrokerJson.keys != newBrokerJson.keys // Broker json changed
-        val shouldUpdateLocalEtags = newBrokerJson.entries != currentBrokerJson.entries // broker state / active state changed
+        val updatedJsons = jsonEtagsFromConfig.toSet() - existingEtags.toSet()
 
-        if (shouldDownloadJson) {
-            val diff = newBrokerJson.keys - currentBrokerJson.keys
-            brokerDataDownloader.downloadBrokerData(diff.map { it.fileName })
-        }
+        pirRepository.updateBrokerJsons(jsonEtagsFromConfig)
 
-        // We update stored json etags once the actual json data have been stored.
-        if (shouldUpdateLocalEtags) {
-            pirRepository.updateBrokerJsons(newBrokerJson)
+        if (updatedJsons.isNotEmpty()) {
+            logcat { "PIR-update: Downloading updated broker json files: $updatedJsons" }
+            brokerDataDownloader.downloadBrokerData(updatedJsons.map { it.fileName })
+        } else {
+            logcat { "PIR-update: No broker json files to update." }
         }
     }
 }

pir/pir-impl/schemas/com.duckduckgo.pir.internal.store.PirDatabase/2.json renamed to pir/pir-impl/schemas/com.duckduckgo.pir.impl.store.PirDatabase/2.json

@@ -75,8 +75,7 @@ class RealPirJobsRunner @Inject constructor(
         val startTimeInMillis = currentTimeProvider.currentTimeMillis()
         emitStartPixel(executionType)
 
-        // This should be all brokers since all brokers have scan steps and inactive brokers are removed
-        val activeBrokers = pirRepository.getAllBrokersForScan().toHashSet()
+        val activeBrokers = pirRepository.getAllActiveBrokers().toHashSet()
 
         // Multiple profile support
         val profileQueries = obtainProfiles()
@@ -88,6 +87,7 @@ class RealPirJobsRunner @Inject constructor(
             emitCompletedPixel(executionType, startTimeInMillis)
             return@withContext Result.success(Unit)
         }
+
         if (activeBrokers.isEmpty()) {
             logcat { "PIR-JOB-RUNNER: No active brokers available. Completing run." }
             pixelSender.reportScanStats(0)
@@ -98,8 +98,19 @@ class RealPirJobsRunner @Inject constructor(
 
         attemptCreateScanJobs(activeBrokers, profileQueries)
         executeScanJobs(context, executionType, activeBrokers)
-        attemptCreateOptOutJobs(activeBrokers)
-        executeOptOutJobs(context)
+
+        val formOptOutBrokers = pirRepository.getBrokersForOptOut(true).toSet()
+        val activeFormOptOutBrokers = formOptOutBrokers.intersect(activeBrokers)
+
+        if (activeFormOptOutBrokers.isEmpty()) {
+            logcat { "PIR-JOB-RUNNER: No active parent brokers available for optout. Completing run." }
+            pixelSender.reportOptOutStats(0)
+            emitCompletedPixel(executionType, startTimeInMillis)
+            return@withContext Result.success(Unit)
+        }
+
+        attemptCreateOptOutJobs(activeFormOptOutBrokers)
+        executeOptOutJobs(context, activeFormOptOutBrokers)
 
         logcat { "PIR-JOB-RUNNER: Completed." }
         emitCompletedPixel(executionType, startTimeInMillis)
@@ -137,7 +148,7 @@ class RealPirJobsRunner @Inject constructor(
         profileQueries: List<ProfileQuery>,
     ) {
         logcat { "PIR-JOB-RUNNER: Attempting to create new scan jobs" }
-        // No scan jobs mean that this is the first time PR is being run OR all jobs have been invalidated.
+        // No scan jobs mean that this is the first time PIR is being run OR all jobs have been invalidated.
         val toCreate = mutableListOf<ScanJobRecord>()
 
         profileQueries.filter {
@@ -187,12 +198,12 @@ class RealPirJobsRunner @Inject constructor(
             }
     }
 
-    private suspend fun attemptCreateOptOutJobs(activeBrokers: Set<String>) {
+    private suspend fun attemptCreateOptOutJobs(activeFormOptOutBrokers: Set<String>) {
         val toCreate = mutableListOf<OptOutJobRecord>()
         val extractedProfiles = pirRepository.getAllExtractedProfiles()
 
         extractedProfiles.forEach {
-            if (activeBrokers.contains(it.brokerName) &&
+            if (activeFormOptOutBrokers.contains(it.brokerName) &&
                 pirSchedulingRepository.getValidOptOutJobRecord(it.dbId) == null
             ) {
                 toCreate.add(
@@ -213,12 +224,13 @@ class RealPirJobsRunner @Inject constructor(
         }
     }
 
-    private suspend fun executeOptOutJobs(context: Context) {
-        val formOptOutBrokers = pirRepository.getBrokersForOptOut(true)
-
+    private suspend fun executeOptOutJobs(
+        context: Context,
+        activeFormOptOutBrokers: Set<String>,
+    ) {
         eligibleOptOutJobProvider.getAllEligibleOptOutJobs(currentTimeProvider.currentTimeMillis())
             .filter {
-                formOptOutBrokers.contains(it.brokerName)
+                activeFormOptOutBrokers.contains(it.brokerName)
             }.also {
                 pixelSender.reportOptOutStats(it.size)
                 if (it.isNotEmpty()) {

pir/pir-impl/schemas/com.duckduckgo.pir.internal.store.PirDatabase/3.json renamed to pir/pir-impl/schemas/com.duckduckgo.pir.impl.store.PirDatabase/3.json

@@ -97,6 +97,7 @@ interface DbpService {
         @field:Json(name = "steps")
         val steps: List<String>,
         val schedulingConfig: PirJsonBrokerSchedulingConfig,
+        val removedAt: Long?,
     )
 
     data class PirJsonBrokerSchedulingConfig(

pir/pir-impl/schemas/com.duckduckgo.pir.internal.store.PirDatabase/4.json renamed to pir/pir-impl/schemas/com.duckduckgo.pir.impl.store.PirDatabase/4.json

@@ -48,7 +48,7 @@ import com.squareup.moshi.Types
 
 @Database(
     exportSchema = true,
-    version = 6,
+    version = 7,
     entities = [
         BrokerJsonEtag::class,
         Broker::class,