Kotlin Coroutines and Flow Explained for Android Developers

class UserRepository(
    private val api: UserApi,                       // Retrofit interface with suspend functions
    private val dao: UserDao,                       // Room DAO with suspend functions
) {
    // Reads top to bottom like synchronous code, but never blocks the caller's thread.
    suspend fun refreshUser(id: String): User {
        val remote = api.getUser(id)                // suspends while waiting for the network
        dao.upsert(remote.toEntity())               // suspends while Room writes on its own executor
        return remote.toDomain()
    }
}

class ReportRepository(
    private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO,   // injectable for tests
) {
    suspend fun exportCsv(file: File, rows: List<Row>) = withContext(ioDispatcher) {
        // Blocking file I/O is safe here; callers can stay on Main.
        file.bufferedWriter().use { writer ->
            rows.forEach { writer.appendLine(it.toCsvLine()) }
        }
    }
}

fun loadFeed() {
    viewModelScope.launch {
        _uiState.update { it.copy(isLoading = true) }
        try {
            // Both requests start immediately; total time ≈ the slower one, not the sum.
            val feed = coroutineScope {
                val posts = async { repository.getPosts() }
                val stories = async { repository.getStories() }
                Feed(posts.await(), stories.await())
            }
            _uiState.update { it.copy(isLoading = false, feed = feed) }
        } catch (e: IOException) {
            // If either request fails, coroutineScope cancels the other one too.
            _uiState.update { it.copy(isLoading = false, error = e.message) }
        }
    }
}

viewModelScope.launch {
    try {
        val result = repository.sync()
        _uiState.update { it.copy(lastSync = result.time) }
    } catch (e: CancellationException) {
        throw e                                   // let cancellation propagate
    } catch (e: Exception) {
        _uiState.update { it.copy(error = "Sync failed") }
    }
}

@Dao
interface NoteDao {
    // Room re-emits automatically whenever the notes table changes.
    @Query("SELECT * FROM notes ORDER BY updatedAt DESC")
    fun observeNotes(): Flow<List<NoteEntity>>
}

@OptIn(FlowPreview::class, ExperimentalCoroutinesApi::class)
class SearchViewModel(private val repository: ProductRepository) : ViewModel() {

    private val query = MutableStateFlow("")

    val results: StateFlow<SearchUiState> = query
        .debounce(300)                                  // wait for typing to pause
        .map { it.trim() }
        .distinctUntilChanged()                         // ignore identical queries
        .flatMapLatest { q ->                           // cancel the previous search when a new query arrives
            if (q.isEmpty()) flowOf(SearchUiState.Idle)
            else flow {
                emit(SearchUiState.Loading)
                emit(SearchUiState.Success(repository.search(q)))
            }.catch { emit(SearchUiState.Error("Search failed")) }
        }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5_000),   // keep alive briefly across rotation
            initialValue = SearchUiState.Idle,
        )

    fun onQueryChange(text: String) { query.value = text }
}

sealed interface SearchUiState {
    data object Idle : SearchUiState
    data object Loading : SearchUiState
    data class Success(val products: List<Product>) : SearchUiState
    data class Error(val message: String) : SearchUiState
}

@Composable
fun SearchRoute(viewModel: SearchViewModel = viewModel()) {
    // Stops collecting when the lifecycle drops below STARTED, resumes when it returns.
    val state by viewModel.results.collectAsStateWithLifecycle()
    SearchScreen(state = state, onQueryChange = viewModel::onQueryChange)
}

viewLifecycleOwner.lifecycleScope.launch {
    viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.results.collect { state -> render(state) }
    }
}

@OptIn(ExperimentalCoroutinesApi::class)
class MainDispatcherRule(
    private val dispatcher: TestDispatcher = UnconfinedTestDispatcher(),
) : TestWatcher() {
    override fun starting(description: Description) = Dispatchers.setMain(dispatcher)
    override fun finished(description: Description) = Dispatchers.resetMain()
}

class SearchViewModelTest {
    @get:Rule val mainRule = MainDispatcherRule()

    private val fakeRepo = object : ProductRepository {
        override suspend fun search(query: String) = listOf(Product("1", "Phone case", 299.0))
    }

    @Test
    fun `typing a query emits success after debounce`() = runTest {
        val vm = SearchViewModel(fakeRepo)
        val states = mutableListOf<SearchUiState>()
        // Collect in the background so stateIn's WhileSubscribed upstream is active.
        backgroundScope.launch(UnconfinedTestDispatcher(testScheduler)) { vm.results.toList(states) }

        vm.onQueryChange("case")
        advanceTimeBy(301)            // move virtual time past the debounce
        runCurrent()

        assertTrue(states.last() is SearchUiState.Success)
    }
}