import kotlinx.coroutines.*

// coroutineScope suspends until ALL children complete.
// If any child fails, all siblings are cancelled.
suspend fun fetchDashboard(): Dashboard = coroutineScope {
    val user = async { userService.getUser() }
    val orders = async { orderService.getRecent() }
    val notifications = async { notificationService.getUnread() }

    Dashboard(
        user = user.await(),
        orders = orders.await(),
        notifications = notifications.await()
    )
}

// supervisorScope lets children fail independently.
// One child's failure does NOT cancel siblings.
suspend fun refreshCaches(): List<Result<Unit>> = supervisorScope {
    val jobs = listOf("users", "products", "inventory").map { cache ->
        async {
            runCatching { cacheService.refresh(cache) }
        }
    }
    jobs.awaitAll()
}

suspend fun processLargeFile(file: File) = coroutineScope {
    file.useLines { lines ->
        lines.forEach { line ->
            // Check cancellation between iterations
            ensureActive()
            processLine(line)
        }
    }
}

// CPU-bound loops MUST check isActive — they won't suspend naturally
suspend fun computeHash(data: ByteArray): String = withContext(Dispatchers.Default) {
    var hash = 0L
    for (i in data.indices) {
        if (!isActive) break  // Respect cancellation
        hash = hash * 31 + data[i]
    }
    hash.toString(16)
}

suspend fun transferFunds(from: Account, to: Account, amount: BigDecimal) {
    try {
        from.debit(amount)
        to.credit(amount)
    } finally {
        // After cancellation, the coroutine is in a "cancelling" state.
        // Suspending calls would throw CancellationException — unless
        // we switch to NonCancellable.
        withContext(NonCancellable) {
            auditLog.record("Transfer attempted: $amount from ${from.id} to ${to.id}")
        }
    }
}

import kotlinx.coroutines.flow.*

// Producing a flow
fun searchResults(query: String): Flow<SearchResult> = flow {
    val page1 = api.search(query, page = 1)
    emit(page1)

    if (page1.hasMore) {
        val page2 = api.search(query, page = 2)
        emit(page2)
    }
}

// Operators: transform, filter, combine
fun processedResults(query: String): Flow<DisplayItem> =
    searchResults(query)
        .filter { it.results.isNotEmpty() }
        .map { page -> page.results.map { it.toDisplayItem() } }
        .flatMapConcat { items -> items.asFlow() }
        .onEach { item -> analytics.trackImpression(item.id) }

// Terminal operator — triggers collection
suspend fun displayResults(query: String) {
    processedResults(query).collect { item ->
        ui.render(item)
    }
}

import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.channels.BufferOverflow

// StateFlow: holds a SINGLE current value, replays latest to new collectors.
// Use for UI state, configuration, or anything with a "current" value.
class CounterViewModel : ViewModel() {
    private val _count = MutableStateFlow(0)
    val count = _count.asStateFlow()

    fun increment() {
        _count.value++
    }
}

// SharedFlow: event stream with configurable replay.
// Use for one-shot events (navigation, toasts, errors).
class EventBus {
    private val _events = MutableSharedFlow<AppEvent>(
        replay = 0,           // Skip replaying old events to new subscribers
        extraBufferCapacity = 64,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )
    val events = _events.asSharedFlow()

    suspend fun emit(event: AppEvent) {
        _events.emit(event)
    }
}

import kotlinx.coroutines.channels.*

// Producer-consumer with produce builder
fun CoroutineScope.produceNumbers(): ReceiveChannel<Int> = produce {
    var n = 1
    while (true) {
        send(n++)
        delay(100)
    }
}

// Fan-out: multiple coroutines consuming from one channel
suspend fun fanOut() = coroutineScope {
    val channel = produceNumbers()

    repeat(3) { workerId ->
        launch {
            for (number in channel) {
                println("Worker $workerId processing $number")
            }
        }
    }
}

// Fan-in: multiple producers writing to one channel
suspend fun fanIn() = coroutineScope {
    val channel = Channel<String>()

    launch { repeat(5) { channel.send("Source A: $it"); delay(100) } }
    launch { repeat(5) { channel.send("Source B: $it"); delay(150) } }

    // Collect all from both sources
    repeat(10) {
        println(channel.receive())
    }
    channel.close()
}

// CoroutineExceptionHandler — last resort for uncaught exceptions in launch.
// Only works with launch (NOT async). Only at root scope level.
val handler = CoroutineExceptionHandler { _, exception ->
    logger.error("Uncaught coroutine exception", exception)
    metrics.incrementCounter("coroutine.unhandled_exception")
}

val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default + handler)

scope.launch {
    // If this throws, handler catches it instead of crashing
    riskyOperation()
}

// For async, exceptions surface at await()
val deferred = scope.async {
    riskyOperation() // Exception stored, not thrown yet
}
try {
    deferred.await() // Exception thrown HERE
} catch (e: Exception) {
    handleError(e)
}

// Wrap individual operations, not the entire coroutine
suspend fun fetchWithFallback(): Data {
    return try {
        remoteApi.fetch()
    } catch (e: IOException) {
        logger.warn("Remote fetch failed, using cache", e)
        localCache.get()
    }
}

// Always rethrow CancellationException — it breaks structured concurrency
suspend fun badExample() {
    try {
        someWork()
    } catch (e: Exception) {
        // Before: This catches CancellationException too!
        // The coroutine won't cancel properly.
    }
}

suspend fun goodExample() {
    try {
        someWork()
    } catch (e: CancellationException) {
        throw e // Always rethrow
    } catch (e: Exception) {
        handleError(e)
    }
}

// Switch dispatcher for blocking calls
suspend fun readConfig(): Config = withContext(Dispatchers.IO) {
    val text = File("config.json").readText() // Blocking call on IO pool
    Json.decodeFromString(text)
}

// Custom dispatcher for limited parallelism
val dbDispatcher = Dispatchers.IO.limitedParallelism(4)

suspend fun queryDatabase(): List<Row> = withContext(dbDispatcher) {
    connection.executeQuery("SELECT * FROM orders")
}

// Before: No lifecycle management, lives until process dies
fun handleRequest(request: Request) {
    GlobalScope.launch {
        auditService.log(request) // If this hangs, it leaks forever
    }
}

// After: Use a scoped coroutine tied to the component lifecycle
class RequestHandler(private val scope: CoroutineScope) {
    fun handleRequest(request: Request) {
        scope.launch {
            auditService.log(request)
        }
    }
}

// After: coroutineScope waits for all children
suspend fun processAll(items: List<Item>) = coroutineScope {
    items.forEach { item ->
        launch { process(item) } // These run concurrently
    }
    // coroutineScope suspends until all children complete
}

// Before: Using a detached scope means no waiting
fun processAllBroken(items: List<Item>) {
    val scope = CoroutineScope(Dispatchers.Default)
    items.forEach { item ->
        scope.launch { process(item) } // No one awaits these!
    }
    // Function returns immediately, work may remain incomplete
}

// Before: Swallowing cancellation breaks the entire coroutine tree
try {
    longRunningWork()
} catch (e: Exception) { /* swallows CancellationException */ }

// After: Explicit rethrow
try {
    longRunningWork()
} catch (e: CancellationException) {
    throw e
} catch (e: Exception) {
    recover(e)
}