refactor: Improve data repository lifecycle safety

app/src/main/kotlin/com/arflix/tv/data/repository/AddonRuntimeAggregator.kt

@@ -11,25 +11,19 @@ class AddonRuntimeAggregator(
         stremioAddons: List<Addon>,
         request: MovieRuntimeRequest
     ): List<StreamSource> {
-        val streams = mutableListOf<StreamSource>()
-        if (stremioAddons.isNotEmpty()) {
-            streams += addonRuntimes[RuntimeKind.STREMIO]
-                ?.resolveMovieStreams(stremioAddons, request)
-                .orEmpty()
-        }
-        return streams
+        if (stremioAddons.isEmpty()) return emptyList()
+        return addonRuntimes[RuntimeKind.STREMIO]
+            ?.resolveMovieStreams(stremioAddons, request)
+            .orEmpty()
     }
 
     suspend fun resolveEpisodeStreams(
         stremioAddons: List<Addon>,
         request: EpisodeRuntimeRequest
     ): List<StreamSource> {
-        val streams = mutableListOf<StreamSource>()
-        if (stremioAddons.isNotEmpty()) {
-            streams += addonRuntimes[RuntimeKind.STREMIO]
-                ?.resolveEpisodeStreams(stremioAddons, request)
-                .orEmpty()
-        }
-        return streams
+        if (stremioAddons.isEmpty()) return emptyList()
+        return addonRuntimes[RuntimeKind.STREMIO]
+            ?.resolveEpisodeStreams(stremioAddons, request)
+            .orEmpty()
     }
 }

app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncCoordinator.kt

@@ -8,6 +8,7 @@ import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.launch
+import java.util.concurrent.atomic.AtomicBoolean
 import javax.inject.Inject
 import javax.inject.Singleton
 
@@ -18,42 +19,48 @@ class CloudSyncCoordinator @Inject constructor(
     private val authRepository: AuthRepository
 ) {
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+    private val lifecycleLock = Any()
     private var collectorJob: Job? = null
     private var flushJob: Job? = null
 
-    @Volatile
-    private var started = false
+    private val started = AtomicBoolean(false)
 
     fun start() {
-        if (started) return
-        started = true
-        collectorJob = scope.launch {
-            invalidationBus.events.collectLatest { invalidation ->
-                if (authRepository.getCurrentUserId().isNullOrBlank()) return@collectLatest
-                cloudSyncRepository.markLocalStateDirty()
-                scheduleFlush(invalidation)
+        synchronized(lifecycleLock) {
+            if (!started.compareAndSet(false, true)) return
+            collectorJob = scope.launch {
+                invalidationBus.events.collectLatest { invalidation ->
+                    if (authRepository.getCurrentUserId().isNullOrBlank()) return@collectLatest
+                    cloudSyncRepository.markLocalStateDirty()
+                    scheduleFlush(invalidation)
+                }
             }
         }
     }
 
     fun stop() {
-        started = false
-        collectorJob?.cancel()
-        flushJob?.cancel()
-        collectorJob = null
-        flushJob = null
+        synchronized(lifecycleLock) {
+            started.set(false)
+            collectorJob?.cancel()
+            flushJob?.cancel()
+            collectorJob = null
+            flushJob = null
+        }
     }
 
     private fun scheduleFlush(invalidation: CloudSyncInvalidation) {
-        flushJob?.cancel()
-        flushJob = scope.launch {
-            delay(debounceMsFor(invalidation.scope))
-            if (authRepository.getCurrentUserId().isNullOrBlank()) return@launch
-            runCatching { cloudSyncRepository.pushToCloud() }
-                .onFailure { error ->
-                    Log.w("CloudSyncCoordinator", "Cloud push failed after ${invalidation.scope}: ${error.message}")
-                    cloudSyncRepository.markLocalStateDirty()
-                }
+        synchronized(lifecycleLock) {
+            if (!started.get()) return
+            flushJob?.cancel()
+            flushJob = scope.launch {
+                delay(debounceMsFor(invalidation.scope))
+                if (authRepository.getCurrentUserId().isNullOrBlank()) return@launch
+                runCatching { cloudSyncRepository.pushToCloud() }
+                    .onFailure { error ->
+                        Log.w("CloudSyncCoordinator", "Cloud push failed after ${invalidation.scope}: ${error.message}")
+                        cloudSyncRepository.markLocalStateDirty()
+                    }
+            }
         }
     }
 

app/src/main/kotlin/com/arflix/tv/data/repository/PlaybackTelemetryRepository.kt

@@ -13,15 +13,17 @@ import javax.inject.Singleton
 class PlaybackTelemetryRepository @Inject constructor(
     @ApplicationContext private val context: Context
 ) {
-    private val startupSamplesKey = longPreferencesKey("telemetry_startup_samples_v1")
-    private val startupAvgMsKey = longPreferencesKey("telemetry_startup_avg_ms_v1")
-    private val startupRetriesKey = longPreferencesKey("telemetry_startup_retries_v1")
-    private val failoverAttemptsKey = longPreferencesKey("telemetry_failover_attempts_v1")
-    private val failoverSuccessesKey = longPreferencesKey("telemetry_failover_successes_v1")
-    private val longRebuffersKey = longPreferencesKey("telemetry_long_rebuffers_v1")
-    private val playbackFailuresKey = longPreferencesKey("telemetry_playback_failures_v1")
-    private val lastStartupMsKey = longPreferencesKey("telemetry_last_startup_ms_v1")
-    private val lastSessionRetriesKey = intPreferencesKey("telemetry_last_session_retries_v1")
+    private companion object {
+        val startupSamplesKey = longPreferencesKey("telemetry_startup_samples_v1")
+        val startupAvgMsKey = longPreferencesKey("telemetry_startup_avg_ms_v1")
+        val startupRetriesKey = longPreferencesKey("telemetry_startup_retries_v1")
+        val failoverAttemptsKey = longPreferencesKey("telemetry_failover_attempts_v1")
+        val failoverSuccessesKey = longPreferencesKey("telemetry_failover_successes_v1")
+        val longRebuffersKey = longPreferencesKey("telemetry_long_rebuffers_v1")
+        val playbackFailuresKey = longPreferencesKey("telemetry_playback_failures_v1")
+        val lastStartupMsKey = longPreferencesKey("telemetry_last_startup_ms_v1")
+        val lastSessionRetriesKey = intPreferencesKey("telemetry_last_session_retries_v1")
+    }
 
     suspend fun recordStartup(startupMs: Long, retries: Int, failoversBeforeStart: Int) {
         val safeStartup = startupMs.coerceAtLeast(0L)