/kotlin-coroutines

name: kotlin-coroutines
description: Advanced Kotlin coroutines patterns for AmethystMultiplatform. Use when working with: (1) Structured concurrency (supervisorScope, coroutineScope), (2) Advanced Flow operators (flatMapLatest, combine, merge, shareIn, stateIn), (3) Channels and callbackFlow, (4) Dispatcher management and context switching, (5) Exception handling (CoroutineExceptionHandler, SupervisorJob), (6) Testing async code (runTest, Turbine), (7) Nostr relay connection pools and subscriptions, (8) Backpressure handling in event streams. Delegates to kotlin-expert for basic StateFlow/SharedFlow patterns. Complements nostr-expert for relay communication.

Kotlin Coroutines - Advanced Async Patterns

Expert guidance for complex async operations in Amethyst: relay pools, event streams, structured concurrency, and testing.

Mental Model

Async Architecture in Amethyst:

Relay Pool (supervisorScope)
    ├── Relay 1 (launch) → callbackFlow → Events
    ├── Relay 2 (launch) → callbackFlow → Events
    └── Relay 3 (launch) → callbackFlow → Events
            ↓
    merge() → distinctBy(id) → shareIn
            ↓
    Multiple Collectors (ViewModels, Services)

**Key principles:**

• **supervisorScope** - Children fail independently
• **callbackFlow** - Bridge callbacks to Flow
• **shareIn/stateIn** - Hot flows from cold
• **Backpressure** - buffer(), conflate(), DROP_OLDEST

When to Use This Skill

Use for **advanced** async patterns:

• Multi-relay subscriptions with supervisorScope
• Complex Flow operators (flatMapLatest, combine, merge)
• callbackFlow for Android callbacks (connectivity, location)
• Backpressure handling in high-frequency streams
• Exception handling with CoroutineExceptionHandler
• Testing coroutines with runTest and Turbine

**Delegate to kotlin-expert for:**

• Basic StateFlow/SharedFlow patterns
• Simple viewModelScope.launch
• MutableStateFlow → asStateFlow()

Core Patterns

Pattern: callbackFlow for Relay Subscriptions

// Real pattern from NostrClientStaticReqAsStateFlow.kt
fun INostrClient.reqAsFlow(
    relay: NormalizedRelayUrl,
    filters: List<Filter>,
): Flow<List<Event>> = callbackFlow {
    val subId = RandomInstance.randomChars(10)
    var hasBeenLive = false
    val eventIds = mutableSetOf<HexKey>()
    var currentEvents = listOf<Event>()

    val listener = object : IRequestListener {
        override fun onEvent(event: Event, ...) {
            if (event.id !in eventIds) {
                currentEvents = if (hasBeenLive) {
                    // After EOSE: prepend
                    listOf(event) + currentEvents
                } else {
                    // Before EOSE: append
                    currentEvents + event
                }
                eventIds.add(event.id)
                trySend(currentEvents)
            }
        }

        override fun onEose(...) {
            hasBeenLive = true
        }
    }

    openReqSubscription(subId, mapOf(relay to filters), listener)

    awaitClose { close(subId) }
}

**Key techniques:** 1. Deduplication with Set 2. EOSE handling (append → prepend strategy) 3. trySend (non-blocking from callback) 4. awaitClose for cleanup

Pattern: Structured Concurrency for Relays

suspend fun connectToRelays(relays: List<Relay>) = supervisorScope {
    relays.forEach { relay ->
        launch {
            try {
                relay.connect()
                relay.subscribe(filters).collect { event ->
                    eventChannel.send(event)
                }
            } catch (e: IOException) {
                Log.e("Relay", "Connection failed: ${relay.url}", e)
                // Other relays continue
            }
        }
    }
}

**Why supervisorScope:**

• One relay failure doesn't cancel others
• All cancelled together when scope cancelled
• Proper cleanup guaranteed

Pattern: Exception Handling for Services

// Real pattern from PushNotificationReceiverService.kt
class MyService : Service() {
    val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
        Log.e("Service", "Caught: ${throwable.message}", throwable)
    }

    private val scope = CoroutineScope(
        Dispatchers.IO + SupervisorJob() + exceptionHandler
    )

    override fun onDestroy() {
        scope.cancel()
        super.onDestroy()
    }
}

**Pattern benefits:**

• SupervisorJob: children fail independently
• ExceptionHandler: log instead of crash
• Scoped lifecycle: cancel all on destroy

Pattern: Network Connectivity as Flow

// Real pattern from ConnectivityFlow.kt
val status = callbackFlow {
    val networkCallback = object : NetworkCallback() {
        override fun onAvailable(network: Network) {
            trySend(ConnectivityStatus.Active(...))
        }
        override fun onLost(network: Network) {
            trySend(ConnectivityStatus.Off)
        }
    }

    connectivityManager.registerCallback(networkCallback)

    // Initial state
    activeNetwork?.let { trySend(ConnectivityStatus.Active(...)) }

    awaitClose {
        connectivityManager.unregisterCallback(networkCallback)
    }
}
    .distinctUntilChanged()
    .debounce(200)  // Stabilize flapping
    .flowOn(Dispatchers.IO)

**Key patterns:** 1. Emit initial state immediately 2. Register callback in flow body 3. Cleanup in awaitClose 4. Stabilize with debounce + distinctUntilChanged

Pattern: Merge Events from Multiple Relays

fun observeFromRelays(
    relays: List<NormalizedRelayUrl>,
    filters: List<Filter>
): Flow<Event> =
    relays.map { relay ->
        client.reqAsFlow(relay, filters)
            .flatMapConcat { it.asFlow() }
    }.merge()
    .distinctBy { it.id }

**Flow:**

• Each relay: `Flow<List<Event>>`
• flatMapConcat: flatten to `Flow<Event>`
• merge(): combine all relays
• distinctBy: deduplicate across relays

Advanced Operators

For comprehensive coverage of Flow operators:

• **flatMapLatest, combine,