fix schedule repository

flow

feature/schedule/src/main/java/ru/fincode/tsudesk/feature/schedule/data/ScheduleRepositoryImpl.kt

@@ -1,10 +1,9 @@
 package ru.fincode.tsudesk.feature.schedule.data
 
-import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.channelFlow
-import kotlinx.coroutines.flow.distinctUntilChanged
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flow
 import ru.fincode.tsudesk.core.common.model.DataResult
 import ru.fincode.tsudesk.core.network.model.NetworkResult
 import ru.fincode.tsudesk.core.network.model.map
@@ -15,7 +14,6 @@ import ru.fincode.tsudesk.feature.schedule.data.mapper.ScheduleNetworkMapper
 import ru.fincode.tsudesk.feature.schedule.domain.model.ScheduleEntity
 import ru.fincode.tsudesk.feature.schedule.domain.model.ScheduleType
 import ru.fincode.tsudesk.feature.schedule.domain.repository.ScheduleRepository
-import java.util.concurrent.atomic.AtomicBoolean
 import javax.inject.Inject
 
 class ScheduleRepositoryImpl @Inject constructor(
@@ -24,47 +22,43 @@ class ScheduleRepositoryImpl @Inject constructor(
     private val mapper: ScheduleNetworkMapper
 ) : ScheduleRepository {
 
-    override fun observeSchedule(type: ScheduleType): Flow<DataResult<ScheduleEntity>> =
-        channelFlow {
-            val key = when (type) {
-                is ScheduleType.Group -> ScheduleCacheKey.group(type.number)
-                is ScheduleType.Teacher -> ScheduleCacheKey.teacher(type.name)
-            }
-
-            // Маркер: "следующий DB-emit считать пришедшим после сети"
-            val markNextAsNetwork = AtomicBoolean(false)
-
-            val cacheJob = launch {
-                local.observeSchedule(key).collect { cached ->
-                    if (cached != null) {
-                        val flag = markNextAsNetwork.getAndSet(false)
-                        trySend(DataResult.Data(cached, refreshedFromNetwork = flag))
-                    }
-                }
-            }
-
-            // refresh: сеть -> БД, без эмита сетевых данных
-            launch {
-                val net: NetworkResult<ScheduleEntity> = when (type) {
-                    is ScheduleType.Group -> remote.loadScheduleByGroup(type.number)
-                        .map(mapper::invoke)
-
-                    is ScheduleType.Teacher -> remote.loadScheduleByTeacher(type.name)
-                        .map(mapper::invoke)
-                }
-
-                when (net) {
-                    is NetworkResult.Success -> {
-                        local.saveSchedule(key, net.data)          // записали
-                        markNextAsNetwork.set(true)               // следующий DB emit пометить
-                    }
-
-                    is NetworkResult.Error -> {
-                        trySend(DataResult.Error(Throwable(net.error.toString())))
-                    }
-                }
-            }
-
-            awaitClose { cacheJob.cancel() }
+    override fun observeSchedule(type: ScheduleType): Flow<DataResult<ScheduleEntity>> = flow {
+        val key = when (type) {
+            is ScheduleType.Group -> ScheduleCacheKey.group(type.number)
+            is ScheduleType.Teacher -> ScheduleCacheKey.teacher(type.name)
         }
+
+        val cached: ScheduleEntity? = local.observeSchedule(key).first()
+        if (cached != null) {
+            emit(DataResult.Data(cached, refreshedFromNetwork = false))
+        }
+
+        val networkResult: NetworkResult<ScheduleEntity> = when (type) {
+            is ScheduleType.Group ->
+                remote.loadScheduleByGroup(type.number).map(mapper::invoke)
+
+            is ScheduleType.Teacher ->
+                remote.loadScheduleByTeacher(type.name).map(mapper::invoke)
+        }
+
+        when (networkResult) {
+            is NetworkResult.Success -> {
+                // Важно: timestamp должен обновляться при записи из сети (либо приходить новым)
+                local.saveSchedule(key, networkResult.data)
+            }
+
+            is NetworkResult.Error -> {
+                emit(DataResult.Error(Throwable(networkResult.error.toString())))
+                return@flow
+            }
+        }
+
+        // 3) DB snapshot после записи — эмитим второй раз (NETWORK)
+        val updated: ScheduleEntity = local.observeSchedule(key).filterNotNull().first()
+
+        // Если не хочешь дублировать, когда данные те же — включи проверку:
+        // if (cached?.timestamp != updated.timestamp) { ... }
+
+        emit(DataResult.Data(updated, refreshedFromNetwork = true))
+    }
 }

feature/schedule/src/main/java/ru/fincode/tsudesk/feature/schedule/data/local/ScheduleLocalDataSourceImpl.kt

@@ -3,6 +3,7 @@ package ru.fincode.tsudesk.feature.schedule.data.local
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.onStart
 import ru.fincode.tsudesk.core.database.schedule.ScheduleDao
 import ru.fincode.tsudesk.feature.schedule.data.datasource.ScheduleLocalDataSource
 import ru.fincode.tsudesk.feature.schedule.data.mapper.toCache
@@ -26,8 +27,11 @@ class ScheduleLocalDataSourceImpl @Inject constructor(
     }
 
     override fun observeSchedule(key: String): Flow<ScheduleEntity?> {
-        return dao.observeSchedule(key)
-            .combine(dao.observeLessons(key)) { schedule, lessons ->
+        val scheduleFlow = dao.observeSchedule(key) // Flow<ScheduleCacheEntity?>
+        val lessonsFlow = dao.observeLessons(key)
+            .onStart { emit(emptyList()) }
+        return scheduleFlow
+            .combine(lessonsFlow) { schedule, lessons ->
                 schedule?.toDomain(lessons)
             }
             .distinctUntilChanged()