amethyst: Skill for Claude Code

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) }
}