refactor: unify db refresh flows

2026-05-14 20:45:24 +03:00
parent f24eae8209
commit f6c6ed59f3
32 changed files with 882 additions and 408 deletions
@@ -16,11 +16,17 @@ import dev.meloda.fast.common.LongPollController
 import dev.meloda.fast.common.VkConstants
 import dev.meloda.fast.common.extensions.listenValue
 import dev.meloda.fast.common.model.LongPollState
+import dev.meloda.fast.data.VkMemoryCache
 import dev.meloda.fast.data.UserConfig
 import dev.meloda.fast.data.processState
 import dev.meloda.fast.datastore.AppSettings
 import dev.meloda.fast.domain.LongPollUpdatesParser
 import dev.meloda.fast.domain.LongPollUseCase
+import dev.meloda.fast.domain.MessagesUseCase
+import dev.meloda.fast.domain.StoreUsersUseCase
+import dev.meloda.fast.model.api.data.VkGroupData
+import dev.meloda.fast.model.api.data.VkUserData
+import dev.meloda.fast.model.api.data.asDomain
 import dev.meloda.fast.model.api.data.LongPollUpdates
 import dev.meloda.fast.model.api.data.VkLongPollData
 import dev.meloda.fast.ui.R
@@ -33,6 +39,7 @@ import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.suspendCancellableCoroutine
+import kotlinx.coroutines.flow.last
 import org.koin.android.ext.android.inject
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.resume
@@ -55,6 +62,8 @@ class LongPollingService : Service() {
    private val coroutineScope = CoroutineScope(coroutineContext)

    private val longPollUseCase: LongPollUseCase by inject()
+    private val messagesUseCase: MessagesUseCase by inject()
+    private val storeUsersUseCase: StoreUsersUseCase by inject()
    private val updatesParser: LongPollUpdatesParser by inject()

    private var currentJob: Job? = null
@@ -150,6 +159,8 @@ class LongPollingService : Service() {
            var serverInfo = getServerInfo()
                ?: throw LongPollException(message = "bad VK response (server info)")

+            syncLongPollHistory(serverInfo)
+
            var lastUpdatesResponse: LongPollUpdates? = getUpdatesResponse(serverInfo)
                ?: throw LongPollException(message = "initiation error: bad VK response (last updates)")

@@ -160,6 +171,7 @@ class LongPollingService : Service() {
                    failCount++
                    serverInfo = getServerInfo()
                        ?: throw LongPollException(message = "failed retrieving server info after error: bad VK response (server info #2)")
+                    syncLongPollHistory(serverInfo)
                    lastUpdatesResponse = getUpdatesResponse(serverInfo)
                    continue
                }
@@ -179,6 +191,7 @@ class LongPollingService : Service() {
                            ?: throw LongPollException(
                                message = "failed retrieving server info after error: bad VK response (server info #3)"
                            )
+                        syncLongPollHistory(serverInfo)
                        lastUpdatesResponse = getUpdatesResponse(serverInfo)
                    }

@@ -196,6 +209,9 @@ class LongPollingService : Service() {
                                updates.forEach(updatesParser::parseNextUpdate)
                            }

+                            AppSettings.LongPoll.ts = lastUpdatesResponse.ts ?: serverInfo.ts
+                            AppSettings.LongPoll.pts = lastUpdatesResponse.pts ?: AppSettings.LongPoll.pts
+
                            lastUpdatesResponse = getUpdatesResponse(serverInfo.copy(ts = newTs))
                        }
                    }
@@ -246,6 +262,73 @@ class LongPollingService : Service() {
        }
    }

+    private suspend fun syncLongPollHistory(serverInfo: VkLongPollData) {
+        val cursorTs = AppSettings.LongPoll.ts ?: serverInfo.ts
+        val cursorPts = AppSettings.LongPoll.pts ?: serverInfo.pts
+        val maxMsgId = messagesUseCase.getLocalMaxMessageId()
+
+        var currentTs = cursorTs
+        var currentPts = cursorPts
+        var more: Int?
+
+        do {
+            val historyResponse = getLongPollHistory(
+                ts = currentTs,
+                pts = currentPts,
+                maxMsgId = maxMsgId
+            ) ?: return
+
+            historyResponse.history.orEmpty().forEach(updatesParser::parseNextUpdate)
+
+            historyResponse.messages?.items.orEmpty().takeIf { it.isNotEmpty() }?.let { rawMessages ->
+                val messages = rawMessages.map { it.asDomain() }
+                messagesUseCase.storeMessages(messages)
+            }
+
+            historyResponse.profiles.orEmpty().takeIf { it.isNotEmpty() }?.let { profiles ->
+                val users = profiles.map(VkUserData::mapToDomain)
+                VkMemoryCache.appendUsers(users)
+                storeUsersUseCase(users).last()
+            }
+
+            historyResponse.groups.orEmpty().takeIf { it.isNotEmpty() }?.let { groups ->
+                VkMemoryCache.appendGroups(groups.map(VkGroupData::mapToDomain))
+            }
+
+            currentTs = historyResponse.ts ?: historyResponse.fromPts ?: currentTs
+            currentPts = historyResponse.newPts ?: historyResponse.pts ?: currentPts
+            more = historyResponse.more
+
+            AppSettings.LongPoll.ts = currentTs
+            AppSettings.LongPoll.pts = currentPts
+        } while (more == 1)
+    }
+
+    private suspend fun getLongPollHistory(
+        ts: Int,
+        pts: Int,
+        maxMsgId: Long?
+    ): dev.meloda.fast.model.api.data.LongPollHistoryResponse? = suspendCancellableCoroutine {
+        longPollUseCase.getLongPollHistory(
+            ts = ts,
+            pts = pts,
+            lpVersion = VkConstants.LP_VERSION,
+            maxMsgId = maxMsgId,
+            eventsLimit = 1000,
+            msgsLimit = 200
+        ).listenValue(coroutineScope) { state ->
+            state.processState(
+                success = { response ->
+                    it.resume(response)
+                },
+                error = { error ->
+                    Log.e(TAG, "getLongPollHistory: error: $error")
+                    it.resume(null)
+                }
+            )
+        }
+    }
+
    private fun handleError(throwable: Throwable) {
        Log.e(TAG, "error: $throwable")

@@ -1,6 +1,7 @@
 package dev.meloda.fast.service.longpolling.di

 import dev.meloda.fast.domain.LongPollUpdatesParser
+import dev.meloda.fast.domain.LongPollUpdatesReducer
 import dev.meloda.fast.domain.LongPollUseCase
 import dev.meloda.fast.domain.LongPollUseCaseImpl
 import org.koin.core.module.dsl.singleOf
@@ -10,4 +11,5 @@ import org.koin.dsl.module
 val longPollModule = module {
    singleOf(::LongPollUseCaseImpl) bind LongPollUseCase::class
    singleOf(::LongPollUpdatesParser)
+    singleOf(::LongPollUpdatesReducer)
 }