Kotlin Flow: Reactive Streams untuk Android Modern

// Cold Flow: tidak melakukan apa-apa sampai ada collector
fun countDown(from: Int): Flow<Int> = flow {
    for (i in from downTo 0) {
        delay(1000) // Delay 1 detik
        emit(i)
        println("Emitted: $i")
    }
}

// Collector pertama → mulai dari awal
// Collector kedua → juga mulai dari awal (independent!)
fun main() = runBlocking {
    val countdown = countDown(5)

    launch {
        countdown.collect { value ->
            println("Collector 1: $value")
        }
    }

    launch {
        delay(2000) // Mulai 2 detik kemudian
        countdown.collect { value ->
            println("Collector 2: $value")
        }
    }
}

// Output:
// Emitted: 4
// Collector 1: 4
// Emitted: 3
// Collector 1: 3
// Emitted: 2
// Collector 1: 2
// Collector 2: 4  ← Collector 2 juga mulai dari 5!
// ...

fun main() = runBlocking {
    val hotFlow = MutableSharedFlow<String>()

    // Collector 1: mendengarkan
    launch {
        hotFlow.collect { value ->
            println("Collector 1: $value")
        }
    }

    delay(1000) // Tunggu collector siap

    // Emit nilai
    hotFlow.emit("Hello")
    hotFlow.emit("World")

    // Collector 2: mulai mendengarkan
    launch {
        hotFlow.collect { value ->
            println("Collector 2: $value")
        }
    }

    delay(500)
    hotFlow.emit("Kotlin Flow") // Kedua collector menerima

    delay(1000)
}

// Output:
// Collector 1: Hello
// Collector 1: World
// Collector 1: Kotlin Flow
// Collector 2: Kotlin Flow  ← Hanya menerima setelah mulai collect

import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow

// ───── Data class untuk UI State ─────
data class UserUiState(
    val isLoading: Boolean = true,
    val userName: String = "",
    val email: String = "",
    val error: String? = null,
    val isLoggedIn: Boolean = false,
)

// ───── ViewModel ─────
class UserViewModel(
    private val userRepository: UserRepository,
) : ViewModel() {

    // Private mutable — hanya bisa diubah dari dalam ViewModel
    private val _uiState = MutableStateFlow(UserUiState())

    // Public immutable — bisa di-collect dari UI
    val uiState: StateFlow<UserUiState> = _uiState.asStateFlow()

    init {
        loadUser()
    }

    fun loadUser() {
        viewModelScope.launch {
            _uiState.value = _uiState.value.copy(isLoading = true, error = null)

            try {
                val user = userRepository.getCurrentUser()
                _uiState.value = _uiState.value.copy(
                    isLoading = false,
                    userName = user.name,
                    email = user.email,
                    isLoggedIn = true,
                )
            } catch (e: Exception) {
                _uiState.value = _uiState.value.copy(
                    isLoading = false,
                    error = e.message ?: "Gagal memuat data",
                )
            }
        }
    }

    fun logout() {
        viewModelScope.launch {
            userRepository.logout()
            _uiState.value = UserUiState() // Reset ke default
        }
    }

    fun updateName(newName: String) {
        _uiState.value = _uiState.value.copy(userName = newName)
    }
}

@Composable
fun UserScreen(viewModel: UserViewModel = hiltViewModel()) {
    // collectAsState() → otomatis recompose saat state berubah
    val uiState by viewModel.uiState.collectAsStateWithLifecycle()

    when {
        uiState.isLoading -> {
            Box(Modifier.fillMaxSize(), contentAlignment = Alignment.Center) {
                CircularProgressIndicator()
            }
        }
        uiState.error != null -> {
            Column(Modifier.fillMaxSize(), verticalArrangement = Arrangement.Center) {
                Text("Error: ${uiState.error}", color = MaterialTheme.colorScheme.error)
                Spacer(Modifier.height(16.dp))
                Button(onClick = { viewModel.loadUser() }) {
                    Text("Coba Lagi")
                }
            }
        }
        else -> {
            Column(Modifier.padding(16.dp)) {
                Text("Halo, ${uiState.userName}!",
                    style = MaterialTheme.typography.headlineMedium)
                Text(uiState.email,
                    style = MaterialTheme.typography.bodyLarge)
                Spacer(Modifier.height(16.dp))
                OutlinedTextField(
                    value = uiState.userName,
                    onValueChange = { viewModel.updateName(it) },
                    label = { Text("Nama") },
                )
                Spacer(Modifier.height(8.dp))
                Button(onClick = { viewModel.logout() }) {
                    Text("Logout")
                }
            }
        }
    }
}

// ── DENGAN LIVEDATA (Lama) ──
class UserViewModel : ViewModel() {
    private val _user = MutableLiveData<User>()
    val user: LiveData<User> = _user

    fun loadUser() {
        viewModelScope.launch {
            _user.value = repository.getUser()
        }
    }
}

// ── DENGAN STATEFLOW (Modern) ──
class UserViewModel : ViewModel() {
    private val _user = MutableStateFlow<User?>(null)
    val user: StateFlow<User?> = _user.asStateFlow()

    fun loadUser() {
        viewModelScope.launch {
            _user.value = repository.getUser()
        }
    }
}

// ── Kapan pakai StateFlow vs LiveData ──
// ✅ StateFlow:
//   - Kotlin Multiplatform project
//   - Butuh operator (map, filter, combine)
//   - Butuh distinctUntilChanged otomatis
//   - Unit testing lebih mudah
//
// ✅ LiveData:
//   - Legacy project yang sudah banyak LiveData
//   - Ingin lifecycle-aware tanpa collectAsStateWithLifecycle()
//   - Tim belum familiar dengan Flow

// ── Event types untuk one-shot events ──
sealed class UserEvent {
    data class NavigateTo(val route: String) : UserEvent()
    data class ShowSnackbar(val message: String) : UserEvent()
    data class ShowToast(val message: String) : UserEvent()
    object NavigateBack : UserEvent()
}

class UserViewModel : ViewModel() {

    // SharedFlow untuk events (one-shot)
    private val _events = MutableSharedFlow<UserEvent>(
        replay = 0,              // Tidak replay event lama
        extraBufferCapacity = 1,  // Buffer 1 event
        onBufferOverflow = BufferOverflow.DROP_OLDEST,
    )
    val events: SharedFlow<UserEvent> = _events.asSharedFlow()

    fun saveUser(name: String) {
        viewModelScope.launch {
            try {
                repository.saveUser(name)
                // Emit event sukses
                _events.emit(UserEvent.ShowSnackbar("Data tersimpan!"))
                _events.emit(UserEvent.NavigateBack)
            } catch (e: Exception) {
                _events.emit(UserEvent.ShowSnackbar("Error: ${e.message}"))
            }
        }
    }
}

@Composable
fun UserEditScreen(
    viewModel: UserViewModel = hiltViewModel(),
    onNavigateBack: () -> Unit,
    onNavigate: (String) -> Unit,
) {
    val snackbarHostState = remember { SnackbarHostState() }

    // Collect events (one-shot) dengan LaunchedEffect
    LaunchedEffect(Unit) {
        viewModel.events.collect { event ->
            when (event) {
                is UserEvent.NavigateTo -> onNavigate(event.route)
                is UserEvent.ShowSnackbar -> {
                    snackbarHostState.showSnackbar(
                        message = event.message,
                        duration = SnackbarDuration.Short,
                    )
                }
                is UserEvent.ShowToast -> {
                    // Toast.makeText(context, event.message, Toast.LENGTH_SHORT).show()
                }
                is UserEvent.NavigateBack -> onNavigateBack()
            }
        }
    }

    Scaffold(
        snackbarHost = { SnackbarHost(snackbarHostState) },
    ) { padding ->
        // ... your UI content
    }
}

// ── map: Transformasi setiap nilai ──
fun getUsers(): Flow<List<User>> = repository.getUserFlow()
    .map { users ->
        users.sortedBy { it.name }
            .filter { it.isActive }
    }

// ── flatMapConcat: Transformasi yang menghasilkan Flow ──
fun getUserWithPosts(): Flow<UserWithPosts> = getUsers()
    .flatMapConcat { user ->
        flow {
            val posts = repository.getPosts(user.id)
            emit(UserWithPosts(user, posts))
        }
    }

// ── filter: Hanya lewatkan nilai yang memenuhi syarat ──
fun getActiveUsers(): Flow<User> = getAllUsers()
    .filter { it.isActive }
    .filter { it.age >= 18 }

// ── distinctUntilChanged: Hanya emit saat nilai berubah ──
fun searchResults(query: Flow<String>): Flow<List<Result>> = query
    .debounce(300)
    .distinctUntilChanged()
    .flatMapLatest { q ->
        repository.search(q)
    }

// ── transform: Transformasi fleksibel (bisa emit 0, 1, atau banyak) ──
fun paginateUsers(pageSize: Int = 20): Flow<List<User>> = flow {
    var page = 0
    var hasMore = true
    while (hasMore) {
        val users = repository.getUsers(page, pageSize)
        if (users.isEmpty()) {
            hasMore = false
        } else {
            emit(users)
            page++
        }
    }
}

// ── scan: Accumulate dengan running result ──
fun runningTotal(numbers: Flow<Int>): Flow<Int> = numbers
    .scan(0) { acc, value -> acc + value }
// 1, 2, 3, 4 → 0, 1, 3, 6, 10

// ── onEach: Side effect tanpa mengubah nilai ──
fun trackedFlow(): Flow<Data> = dataFlow
    .onEach { analytics.trackEvent("data_received", it) }
    .onEach { println("Data received: $it") }

// ── combine: Gabungkan 2+ flow, emit saat ada yang berubah ──
val searchState: Flow<SearchUiState> = combine(
    searchQuery,     // Flow<String>
    searchResults,   // Flow<List<Result>>
    isLoading,       // Flow<Boolean>
) { query, results, loading ->
    SearchUiState(
        query = query,
        results = results,
        isLoading = loading,
        isEmpty = results.isEmpty() && !loading,
    )
}

// ── zip: Gabungkan 2 flow, emit saat KEDUANYA punya nilai baru ──
val userProfile: Flow<UserProfile> = zip(
    userFlow,    // Flow<User>
    avatarFlow,  // Flow<Avatar>
) { user, avatar ->
    UserProfile(user, avatar)
}

// ── merge: Gabungkan 2+ flow, semua emit di-forward ──
val allEvents: Flow<Event> = merge(
    networkEvents,
    databaseEvents,
    localEvents,
)

// ── flatMapLatest: Batalkan flow lama saat nilai baru datang ──
// Cocok untuk search — batalkan request lama saat user mengetik
fun searchUsers(query: Flow<String>): Flow<List<User>> = query
    .debounce(300)
    .flatMapLatest { q ->
        if (q.isBlank()) flowOf(emptyList())
        else repository.searchUsers(q)
    }

// ── collect: Collect semua nilai ──
viewModelScope.launch {
    dataFlow.collect { value ->
        println(value)
    }
}

// ── first: Ambil nilai pertama ──
val firstUser = usersFlow.first()

// ── toList: Collect semua ke list ──
val allUsers = usersFlow.toList()

// ── reduce: Accumulate ke satu nilai ──
val totalPrice = itemsFlow.reduce { acc, item ->
    acc.copy(total = acc.total + item.price)
}

// ── fold: Reduce dengan initial value ──
val sum = numbersFlow.fold(0) { acc, num -> acc + num }

// ── single: Ambil satu-satunya nilai (error jika lebih) ──
val config = configFlow.single()

// ── stateIn: Convert cold flow ke StateFlow ──
class UserViewModel : ViewModel() {
    // Cold flow → StateFlow (lifecycle-aware)
    val users: StateFlow<List<User>> = repository
        .getUserFlow()
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5000), // Stop 5s setelah no subscriber
            initialValue = emptyList(),
        )
}

// ── shareIn: Convert cold flow ke SharedFlow ──
val events: SharedFlow<Event> = eventSource
    .shareIn(
        scope = viewModelScope,
        started = SharingStarted.Eagerly, // Mulai langsung
        replay = 0,
    )

// ── catch operator: Tangkap error di dalam flow chain ──
val safeDataFlow: Flow<Result<Data>> = dataFlow
    .map { data -> Result.success(data) }
    .catch { e ->
        // Tangkap semua exception dari upstream
        emit(Result.failure(e))
        // Atau emit fallback value
        // emit(Result.success(defaultData))
    }

// ── try-catch di collect ──
viewModelScope.launch {
    try {
        dataFlow.collect { value ->
            // proses value
        }
    } catch (e: Exception) {
        _uiState.value = _uiState.value.copy(error = e.message)
    }
}

// ── onErrorReturn pattern (seperti RxJava onErrorReturn) ──
fun safeUsers(): Flow<List<User>> = repository.getUserFlow()
    .catch { e ->
        println("Error fetching users: ${e.message}")
        emit(emptyList()) // Fallback ke list kosong
    }

// ── retry: Coba lagi setelah error ──
fun resilientUsers(): Flow<List<User>> = repository.getUserFlow()
    .retry(3) { cause ->
        // Retry hanya untuk IOException
        cause is IOException
    }
    .catch { e ->
        emit(emptyList())
    }

// ── retryWhen: Retry dengan exponential backoff ──
fun resilientFlow(): Flow<Data> = dataFlow
    .retryWhen { cause, attempt ->
        if (cause is IOException && attempt < 3) {
            delay(1000L * (attempt + 1)) // Exponential backoff
            true // Retry
        } else {
            false // Jangan retry
        }
    }
    .catch { e ->
        emit(Data.EMPTY)
    }

@HiltViewModel
class ProductViewModel @Inject constructor(
    private val productRepository: ProductRepository,
    private val cartRepository: CartRepository,
) : ViewModel() {

    // ── Search Query ──
    private val _searchQuery = MutableStateFlow("")
    val searchQuery: StateFlow<String> = _searchQuery.asStateFlow()

    // ── Category Filter ──
    private val _selectedCategory = MutableStateFlow<Category?>(null)
    val selectedCategory: StateFlow<Category?> = _selectedCategory.asStateFlow()

    // ── Products (derived dari query + category) ──
    val products: StateFlow<List<Product>> = combine(
        _searchQuery.debounce(300).distinctUntilChanged(),
        _selectedCategory,
    ) { query, category ->
        Pair(query, category)
    }.flatMapLatest { (query, category) ->
        productRepository.getProducts(
            query = query.ifBlank { null },
            category = category,
        )
    }.catch { e ->
        emit(emptyList())
    }.stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5000),
        initialValue = emptyList(),
    )

    // ── UI State ──
    val uiState: StateFlow<ProductUiState> = combine(
        products,
        _searchQuery,
        _selectedCategory,
        cartRepository.cartItems,
    ) { products, query, category, cartItems ->
        ProductUiState(
            products = products,
            searchQuery = query,
            selectedCategory = category,
            cartItemCount = cartItems.size,
            isLoading = false,
        )
    }.stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5000),
        initialValue = ProductUiState(isLoading = true),
    )

    // ── Events ──
    private val _events = MutableSharedFlow<ProductEvent>()
    val events: SharedFlow<ProductEvent> = _events.asSharedFlow()

    // ── Actions ──
    fun updateSearch(query: String) {
        _searchQuery.value = query
    }

    fun selectCategory(category: Category?) {
        _selectedCategory.value = category
    }

    fun addToCart(product: Product) {
        viewModelScope.launch {
            try {
                cartRepository.addToCart(product)
                _events.emit(ProductEvent.ShowSnackbar(
                    "${product.name} ditambahkan ke keranjang"
                ))
            } catch (e: Exception) {
                _events.emit(ProductEvent.ShowSnackbar(
                    "Gagal menambahkan: ${e.message}"
                ))
            }
        }
    }
}

data class ProductUiState(
    val products: List<Product> = emptyList(),
    val searchQuery: String = "",
    val selectedCategory: Category? = null,
    val cartItemCount: Int = 0,
    val isLoading: Boolean = false,
)

sealed class ProductEvent {
    data class ShowSnackbar(val message: String) : ProductEvent()
    data class NavigateTo(val route: String) : ProductEvent()
}

@OptIn(ExperimentalMaterial3Api::class)
@Composable
fun ProductScreen(
    viewModel: ProductViewModel = hiltViewModel(),
    onNavigate: (String) -> Unit,
) {
    val uiState by viewModel.uiState.collectAsStateWithLifecycle()
    val searchQuery by viewModel.searchQuery.collectAsStateWithLifecycle()
    val snackbarHostState = remember { SnackbarHostState() }

    // Collect events
    LaunchedEffect(Unit) {
        viewModel.events.collect { event ->
            when (event) {
                is ProductEvent.ShowSnackbar -> {
                    snackbarHostState.showSnackbar(event.message)
                }
                is ProductEvent.NavigateTo -> onNavigate(event.route)
            }
        }
    }

    Scaffold(
        topBar = {
            TopAppBar(
                title = { Text("Produk") },
                actions = {
                    BadgedBox(badge = {
                        if (uiState.cartItemCount > 0) {
                            Badge { Text("${uiState.cartItemCount}") }
                        }
                    }) {
                        IconButton(onClick = { onNavigate("/cart") }) {
                            Icon(Icons.Default.ShoppingCart, "Keranjang")
                        }
                    }
                },
            )
        },
        snackbarHost = { SnackbarHost(snackbarHostState) },
    ) { padding ->
        Column(Modifier.padding(padding)) {
            // Search Bar
            SearchBar(
                query = searchQuery,
                onQueryChange = { viewModel.updateSearch(it) },
                onSearch = {},
                active = false,
                onActiveChange = {},
                modifier = Modifier.fillMaxWidth().padding(horizontal = 16.dp),
                placeholder = { Text("Cari produk...") },
                leadingIcon = { Icon(Icons.Default.Search, "Cari") },
            ) {}

            // Category Chips
            CategoryChips(
                selectedCategory = uiState.selectedCategory,
                onSelectCategory = { viewModel.selectCategory(it) },
            )

            // Product List
            if (uiState.isLoading) {
                Box(Modifier.fillMaxSize(), contentAlignment = Alignment.Center) {
                    CircularProgressIndicator()
                }
            } else {
                LazyColumn(Modifier.fillMaxSize()) {
                    items(uiState.products, key = { it.id }) { product ->
                        ProductCard(
                            product = product,
                            onAddToCart = { viewModel.addToCart(product) },
                        )
                    }
                }
            }
        }
    }
}

// ── Network + Database (single source of truth) ──
class ProductRepository @Inject constructor(
    private val api: ProductApi,
    private val dao: ProductDao,
) {
    // Data selalu dari database (offline-first)
    // Network hanya untuk sync
    fun getProducts(
        query: String? = null,
        category: Category? = null,
    ): Flow<List<Product>> = flow {
        // 1. Emit data dari database dulu (cache)
        val cachedProducts = dao.getProducts(query, category)
        emit(cachedProducts)

        // 2. Fetch dari network
        try {
            val networkProducts = api.getProducts(query, category)
            // 3. Simpan ke database
            dao.upsertAll(networkProducts.map { it.toEntity() })
        } catch (e: Exception) {
            // Network error — data cache sudah di-emit
            println("Network error: ${e.message}")
        }

        // 4. Emit data terbaru dari database (setelah update)
        val freshProducts = dao.getProducts(query, category)
        emit(freshProducts)
    }.distinctUntilChanged()

    // ── Room DAO dengan Flow ──
    @Dao
    interface ProductDao {
        @Query("SELECT * FROM products WHERE (:query IS NULL OR name LIKE '%' || :query || '%') AND (:category IS NULL OR category = :category)")
        suspend fun getProducts(query: String?, category: Category?): List<Product>

        // Room mendukung Flow langsung — auto-update saat data berubah
        @Query("SELECT * FROM products")
        fun observeProducts(): Flow<List<ProductEntity>>

        @Upsert
        suspend fun upsertAll(products: List<ProductEntity>)
    }
}

// ── Alternatif: Room Flow langsung ──
class OfflineFirstProductRepository @Inject constructor(
    private val api: ProductApi,
    private val dao: ProductDao,
) {
    // Database sebagai source of truth — otomatis update
    fun getProducts(): Flow<List<Product>> = dao.observeProducts()
        .map { entities -> entities.map { it.toDomain() } }

    // Sync: fetch dari network, update ke database
    suspend fun refresh() {
        try {
            val products = api.getProducts()
            dao.upsertAll(products.map { it.toEntity() })
        } catch (e: Exception) {
            throw RefreshFailedException(e)
        }
    }
}

// dependencies: testImplementation("app.cash.turbine:turbine:1.0.0")

@Test
fun `products flow emits empty list initially`() = runTest {
    // Given
    val repository = FakeProductRepository()
    val viewModel = ProductViewModel(repository)

    // When & Then
    viewModel.uiState.test {
        val initialState = awaitItem()
        assertEquals(true, initialState.isLoading)

        val loadedState = awaitItem()
        assertEquals(false, loadedState.isLoading)
        assertEquals(emptyList(), loadedState.products)
    }
}

@Test
fun `search filters products`() = runTest {
    // Given
    val repository = FakeProductRepository(
        products = listOf(
            Product("1", "Laptop", Category.ELECTRONICS),
            Product("2", "Phone", Category.ELECTRONICS),
            Product("3", "T-Shirt", Category.CLOTHING),
        )
    )
    val viewModel = ProductViewModel(repository)

    viewModel.uiState.test {
        // Skip initial loading state
        skipItems(2)

        // When: search for "lap"
        viewModel.updateSearch("lap")

        // Then
        val state = awaitItem()
        assertEquals(1, state.products.size)
        assertEquals("Laptop", state.products[0].name)
    }
}

@Test
fun `combine operators work correctly`() = runTest {
    val flow1 = flowOf(1, 2, 3)
    val flow2 = flowOf("A", "B", "C")

    combine(flow1, flow2) { a, b -> "$a$b" }.test {
        // combine emits when any flow emits
        // First: 1+A="1A"
        assertEquals("1A", awaitItem())
        // Then 2 comes: 2+A="2A"
        assertEquals("2A", awaitItem())
        // Then B comes: 2+B="2B"
        assertEquals("2B", awaitItem())
        // Then 3 comes: 3+B="3B"
        assertEquals("3B", awaitItem())
        // Then C comes: 3+C="3C"
        assertEquals("3C", awaitItem())
        awaitComplete()
    }
}

@Test
fun `search debounce works`() = runTest {
    // Given
    val queryFlow = MutableSharedFlow<String>()
    val results = mutableListOf<List<String>>()

    val searchFlow = queryFlow
        .debounce(300)
        .distinctUntilChanged()
        .flatMapLatest { query ->
            flowOf(listOf("Result for: $query"))
        }

    // When: Collect in background
    val job = launch(UnconfinedTestDispatcher()) {
        searchFlow.collect { results.add(it) }
    }

    // Emit rapid queries
    queryFlow.emit("a")
    queryFlow.emit("ab")
    queryFlow.emit("abc") // Only this should emit (debounce)

    // Advance time to trigger debounce
    advanceTimeBy(400)

    // Then
    assertEquals(1, results.size)
    assertEquals(listOf("Result for: abc"), results[0])

    job.cancel()
}

// ── Using toList() ──
@Test
fun `flow emits expected values`() = runTest {
    val flow = flow {
        emit(1)
        emit(2)
        emit(3)
    }

    val result = flow.toList()
    assertEquals(listOf(1, 2, 3), result)
}

// ── Using first() ──
@Test
fun `flow first value`() = runTest {
    val flow = flow {
        emit("Hello")
        emit("World")
    }

    assertEquals("Hello", flow.first())
}