Fix stream pill flicker when returning to the foreground (#1087)

Co-authored-by: Aleksandar Ilic <aleksandar@appollo41.com>

Author: akurcubic

app/src/main/kotlin/net/primal/android/user/subscriptions/SubscriptionsManager.kt

@@ -29,6 +29,7 @@ import net.primal.core.networking.primal.PrimalApiClient
 import net.primal.core.networking.primal.PrimalCacheFilter
 import net.primal.core.networking.primal.PrimalSocketSubscription
 import net.primal.core.utils.coroutines.DispatcherProvider
+import net.primal.core.utils.runCatching
 import net.primal.core.utils.serialization.encodeToJsonString
 import net.primal.data.remote.api.notifications.model.PubkeyRequestBody
 import net.primal.domain.streams.StreamRepository
@@ -138,6 +139,7 @@ class SubscriptionsManager @Inject constructor(
 
     private fun launchStreamsFromFollowsSubscription(userId: String) =
         scope.launch {
+            runCatching { streamRepository.fetchLiveEventsFromFollows(userId = userId) }
             streamRepository.startLiveEventsFromFollowsSubscription(userId = userId)
         }
 

data/caching/remote/src/commonMain/kotlin/net/primal/data/remote/api/stream/LiveStreamApi.kt

@@ -17,5 +17,7 @@ interface LiveStreamApi {
 
     suspend fun subscribeToLiveEventsFromFollows(userId: String): Flow<NostrEvent>
 
+    suspend fun getLiveEventsFromFollowsSnapshot(userId: String): List<NostrEvent>
+
     suspend fun findLiveStream(body: FindLiveStreamRequestBody): FindLiveStreamResponse
 }

data/caching/remote/src/commonMain/kotlin/net/primal/data/remote/api/stream/LiveStreamApiImpl.kt

@@ -2,11 +2,16 @@ package net.primal.data.remote.api.stream
 
 import io.github.aakira.napier.Napier
 import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
 import kotlin.uuid.Uuid
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.mapNotNull
+import kotlinx.coroutines.flow.takeWhile
+import kotlinx.coroutines.flow.timeout
 import net.primal.core.networking.primal.PrimalApiClient
 import net.primal.core.networking.primal.PrimalCacheFilter
 import net.primal.core.networking.sockets.NostrIncomingMessage
@@ -73,6 +78,36 @@ class LiveStreamApiImpl(
             }
     }
 
+    @OptIn(FlowPreview::class)
+    override suspend fun getLiveEventsFromFollowsSnapshot(userId: String): List<NostrEvent> {
+        val subscriptionId = Uuid.random().toPrimalSubscriptionId()
+        val liveEvents = mutableListOf<NostrEvent>()
+        try {
+            primalApiClient
+                .subscribe(
+                    subscriptionId = subscriptionId,
+                    message = PrimalCacheFilter(
+                        primalVerb = PrimalVerb.LIVE_EVENTS_FROM_FOLLOWS.id,
+                        optionsJson = LiveEventsFromFollowsRequest(pubkey = userId).encodeToJsonString(),
+                    ),
+                )
+                .takeWhile {
+                    it is NostrIncomingMessage.EventMessage || it is NostrIncomingMessage.EventsMessage
+                }
+                .timeout(5.seconds)
+                .collect { message ->
+                    when (message) {
+                        is NostrIncomingMessage.EventMessage -> message.nostrEvent?.let { liveEvents.add(it) }
+                        is NostrIncomingMessage.EventsMessage -> liveEvents.addAll(message.nostrEvents)
+                        else -> Unit
+                    }
+                }
+        } catch (error: TimeoutCancellationException) {
+            Napier.w(throwable = error) { "Timed out collecting `live_events_from_follows` snapshot." }
+        }
+        return liveEvents.filter { it.kind == NostrEventKind.LiveActivity.value }
+    }
+
     override suspend fun findLiveStream(body: FindLiveStreamRequestBody): FindLiveStreamResponse {
         val queryResult = primalApiClient.query(
             message = PrimalCacheFilter(

data/caching/repository/src/commonMain/kotlin/net/primal/data/repository/streams/StreamRepositoryImpl.kt

@@ -112,9 +112,35 @@ class StreamRepositoryImpl(
             return@withContext job
         }
 
+    override suspend fun fetchLiveEventsFromFollows(userId: String) =
+        withContext(dispatcherProvider.io()) {
+            val currentLiveStreams = liveStreamApi.getLiveEventsFromFollowsSnapshot(userId = userId)
+                .mapNotNull { it.asStreamData() }
+
+            currentLiveStreams.forEach { liveActivity ->
+                scope.launch {
+                    profileRepository.fetchMissingProfiles(profileIds = listOf(liveActivity.mainHostId))
+                }
+            }
+
+            database.withTransaction {
+                database.streams().upsertStreamData(data = currentLiveStreams)
+                database.streamFollows().deleteAllByOwnerId(ownerId = userId)
+                currentLiveStreams
+                    .filter { it.isLive() }
+                    .forEach { liveActivity ->
+                        database.streamFollows().upsert(
+                            data = StreamFollowsCrossRef(
+                                streamATag = liveActivity.aTag,
+                                ownerId = userId,
+                            ),
+                        )
+                    }
+            }
+        }
+
     override suspend fun startLiveEventsFromFollowsSubscription(userId: String) =
         withContext(dispatcherProvider.io()) {
-            database.streamFollows().deleteAllByOwnerId(ownerId = userId)
             liveStreamApi.subscribeToLiveEventsFromFollows(userId = userId)
                 .collect { liveActivityEvent ->
                     val liveActivity = liveActivityEvent.asStreamData() ?: return@collect

domain/primal/src/commonMain/kotlin/net/primal/domain/streams/StreamRepository.kt

@@ -35,6 +35,8 @@ interface StreamRepository {
         streamContentModerationMode: StreamContentModerationMode,
     ): Job
 
+    suspend fun fetchLiveEventsFromFollows(userId: String)
+
     suspend fun startLiveEventsFromFollowsSubscription(userId: String)
 
     fun observeLiveEventsFromFollows(userId: String): Flow<List<Stream>>