Article: Beyond CLEAN and MVP: Architecting an Offline-first Reactive Data Layer in Android

package com.example.healthtracker.data.heartrate.api.model

import java.time.Instant

data class HeartRateRecord(
    val id: String,
    val bpm: Int,
    val timestamp: Instant
)

package com.example.healthtracker.data.heartrate.api

import com.example.healthtracker.data.heartrate.api.model.HeartRateRecord
import kotlinx.coroutines.flow.Flow
import java.time.Instant

interface HeartRateRepository {
    /**
     * 返回心率记录的响应式流。
     * 如果本地缓存过期，将在后台触发网络刷新。
     */
    fun observeHeartRates(start: Instant, end: Instant): Flow<List<HeartRateRecord>>

    /**
     * 上传新的心率记录。直到服务器确认后才会挂起。
     */
    suspend fun upsertHeartRates(records: List<HeartRateRecord>)
}

package com.example.healthtracker.data.core.caching

import java.time.Instant

data class Cached<out T>(
    val value: T,
    val insertionTime: Instant
)

package com.example.healthtracker.data.heartrate.impl

import com.example.healthtracker.data.heartrate.api.HeartRateRepository
import com.example.healthtracker.data.heartrate.api.model.HeartRateRecord
import com.example.healthtracker.data.heartrate.impl.local.HeartRateLocalDataSource
import com.example.healthtracker.data.heartrate.impl.remote.HeartRateRemoteDataSource
import com.example.healthtracker.data.core.caching.Cached
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.launch
import java.time.Duration
import java.time.Instant
import kotlin.coroutines.CoroutineContext

internal class HeartRateFetchAndStoreRepository(
    private val localDS: HeartRateLocalDataSource,
    private val remoteDS: HeartRateRemoteDataSource,
    private val appScope: CoroutineScope,
    private val lightweightContext: CoroutineContext,
    private val cacheTtl: Duration = Duration.ofMinutes(10)
) : HeartRateRepository {

    override fun observeHeartRates(start: Instant, end: Instant): Flow<List<HeartRateRecord>> {
        return localDS.readHeartRates(start, end)
            .onStart {
                // Asynchronous background refresh execution
                appScope.launch { triggerRefreshIfNeeded(start, end) }
            }
            .map { cachedList -> cachedList.map { it.value } }
            .flowOn(lightweightContext)
    }

    private suspend fun triggerRefreshIfNeeded(start: Instant, end: Instant) {
        val cachedData = localDS.readHeartRatesOnce(start, end)
        if (cachedData.isEmpty() || isStale(cachedData)) {
            try {
                val remoteData = remoteDS.fetchHeartRates(start, end)
                localDS.writeHeartRates(remoteData)
            } catch (e: Exception) {
                // Fail silently; UI continues displaying cachedData
            }
        }
    }

    private fun isStale(data: List<Cached<HeartRateRecord>>): Boolean {
        val oldestAllowed = Instant.now().minus(cacheTtl)
        return data.any { it.insertionTime.isBefore(oldestAllowed) }
    }

    override suspend fun upsertHeartRates(records: List<HeartRateRecord>) {
        // Synchronous Mutation: Server write must succeed before updating DB
        val serverConfirmed = remoteDS.uploadHeartRates(records)
        localDS.writeHeartRates(serverConfirmed)
    }
}

interface HeartRateLocalDataSource {
    fun readHeartRates(start: Instant, end: Instant): Flow<List<Cached<HeartRateRecord>>>
    suspend fun readHeartRatesOnce(start: Instant, end: Instant): List<Cached<HeartRateRecord>>
    suspend fun writeHeartRates(records: List<HeartRateRecord>)
}

数据库实体（HeartRateEntity.kt）
@Entity(tableName = "heart_rate_records")
data class HeartRateEntity(
    @PrimaryKey val id: String,
    val bpm: Int,
    val timestamp: Instant,
    val insertionTime: Instant
) {
    fun toModel() = HeartRateRecord(id = id, bpm = bpm, timestamp = timestamp)
}

internal class HeartRateRoomDataSource(
    private val dao: HeartRateDao,
    private val lightweightContext: CoroutineContext
) : HeartRateLocalDataSource {

    override fun readHeartRates(start: Instant, end: Instant): Flow<List<Cached<HeartRateRecord>>> {
        return dao.observeHeartRates(start, end)
            .distinctUntilChanged() // 防止 Room 重复发送数据
            .map { entities ->
                entities.map { Cached(it.toModel(), it.insertionTime) }
            }
            .flowOn(lightweightContext)
    }

    override suspend fun readHeartRatesOnce(start: Instant, end: Instant): List<Cached<HeartRateRecord>> {
        return dao.getHeartRates(start, end).map { Cached(it.toModel(), it.insertionTime) }
    }

    override suspend fun writeHeartRates(records: List<HeartRateRecord>) {
        val entities = records.map { 
            HeartRateEntity(it.id, it.bpm, it.timestamp, Instant.now()) 
        }
        dao.insertOrUpdate(entities)
    }
}

@HiltViewModel
class HeartRateViewModel @Inject constructor(
    private val repository: HeartRateRepository
) : ViewModel() {

    // 持久状态流
    val uiState: StateFlow<UiState> = repository
        .observeHeartRates(Instant.now().minus(1, ChronoUnit.DAYS), Instant.now())
        .map { records -> UiState.Success(records) }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5000),
            initialValue = UiState.Loading
        )

    // 短暂事件流
    private val _faultEvents = MutableSharedFlow<FaultEvent>(
        replay = 1,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )
    val faultEvents = _faultEvents.asSharedFlow()
}

override fun readHeartRates(start: Instant, end: Instant): Flow<List<Cached<HeartRateRecord>>> {
    val savedRecordsFlow = dao.observeHeartRates(start, end)
    val pendingMutationsFlow = mutationDao.observePendingAddMutations()

    return savedRecordsFlow.combine(pendingMutationsFlow) { saved, mutations ->
        val mergedList = saved.map { Cached(it.toModel(), it.insertionTime) }.toMutableList()
        
        mutations.forEach { mutation ->
            if (mutation.timestamp in start..end) {
                // 在列表上叠加本地待处理的修改
                mergedList.removeAll { it.value.id == mutation.localId }
                mergedList.add(
                    Cached(
                        value = HeartRateRecord(mutation.localId, mutation.bpm, mutation.timestamp),
                        insertionTime = mutation.localCreationTime
                    )
                )
            }
        }
        mergedList.sortedByDescending { it.value.timestamp }
    }.flowOn(lightweightContext)
}

@VisibleForTesting
interface HeartRateRepositoryTestExtensions {
    suspend fun seedLocalHeartRates(records: List<HeartRateRecord>)
    suspend fun clearLocalCache()
}

internal class HeartRateFetchAndStoreRepository(
    private val localDS: HeartRateLocalDataSource,
    // ...
) : HeartRateRepository, HeartRateRepositoryTestExtensions {
    
    // 仓库实现...

    override suspend fun seedLocalHeartRates(records: List<HeartRateRecord>) {
        localDS.writeHeartRates(records)
    }

    override suspend fun clearLocalCache() {
        localDS.clearAll()
    }
}

@HiltAndroidTest
@RunWith(AndroidJUnit4::class)
@Config(application = HiltTestApplication::class)
class HeartRateRepositoryTest {

    @get:Rule val hiltRule = HiltAndroidRule(this)

    @Inject lateinit var repository: HeartRateRepository
    @Inject lateinit var testExtensions: HeartRateRepositoryTestExtensions

    @Before
    fun setUp() {
        hiltRule.inject()
    }

    @Test
    fun observeHeartRates_emitsSeededData() = runTest {
        val now = Instant.now()
        val records = listOf(HeartRateRecord("1", 72, now))
        

        // 通过测试扩展直接向数据库添加数据
        testExtensions.seedLocalHeartRates(records)

        val flow = repository.observeHeartRates(now.minusSeconds(60), now.plusSeconds(60))
        

        assertThat(flow.first()).containsExactlyElementsIn(records)
    }
}