Kotlin Flow in Muse: Connecting AccountManager to ElevenLabs Providers

Kotlin Flow acts as the reactive bridge between AccountManager and ElevenLabs-based providers, automatically triggering client initialization whenever the API key updates in DataStore without manual polling or refresh calls.

The Muse repository demonstrates a reactive architecture that decouples credential persistence from service activation. By exposing DataStore values as cold streams, AccountManager enables the ElevenLabProcessor to lazily initialize and reconfigure itself whenever users modify their ElevenLabs API key or subscription status.

Exposing Stored Values as Flows in AccountManager

In AccountManager.kt, user credentials and subscription data are exposed as Flow instances that emit whenever the underlying DataStore changes. This reactive pattern eliminates the need for manual refresh mechanisms in the UI layer.

The API key is exposed as a nullable string flow:

var apiKey: Flow<String?> = dataStore.data.map { it[key] }
// Located in AccountManager.kt, lines 17-20

Similarly, the subscription status maps stored strings to enum values:

var subscriptionStatus: Flow<SubscriptionStatus?> = dataStore.data
    .map { it[statusKey]?.let { SubscriptionStatus.valueOf(it) } }
// Located in AccountManager.kt, lines 28-32

These flows emit new values immediately when the DataStore updates, such as when a user saves a new API key through the settings screen.

Lazy Provider Initialization via Flow Collection

The ElevenLabProcessor class implements all provider interfaces (TTSProvider, AudioIsolationProvider, SoundEffectProvider, STTProvider) and subscribes to the apiKey flow to manage the ElevenLabs client lifecycle. This ensures the client only initializes when valid credentials exist and automatically recreates when keys rotate.

The Initialization Sequence

The initialize() function in ElevenLabProcessor.kt (lines 49-63) demonstrates reactive client creation:

private fun initialize() {
    if (!isInitialized) {
        initJob?.cancel()
        initJob = scope.launch {
            accountManager.apiKey
                .onStart { isInitialized = true }
                .distinctUntilChanged()
                .collectLatest { apiKey ->
                    if (apiKey != null) client = ElevenLabsClient(apiKey)
                }
        }
    }
}

onStart immediately marks the processor as initializing to prevent redundant launch attempts. distinctUntilChanged filters duplicate emissions, ensuring the client only recreates when the key actually changes rather than on every DataStore write. collectLatest cancels any in-progress initialization when a new key arrives, always favoring the most recent credential.

Handling Uninitialized Clients with Retry Logic

Provider operations (generate, removeBackgroundNoise, makeSoundEffects, transcribeAudio) delegate to requireClient(), which implements a retry-based handshake for flow-driven initialization. Located in ElevenLabProcessor.kt (lines 67-80), this function pauses execution briefly to allow the flow collection to complete:

private suspend fun requireClient(retryCount: Int = 2): Result<ElevenLabsClient> {
    return if (!::client.isInitialized) {
        if (!isInitialized && retryCount > 0) {
            initialize()
            delay(200)
            requireClient(retryCount - 1)
        } else {
            Result.failure(IllegalStateException("apiKey is not set"))
        }
    } else {
        Result.success(client)
    }
}

If the client remains uninitialized, the function triggers initialize(), waits 200 milliseconds, then recursively retries up to two times. This graceful degradation ensures that providers block briefly for credentials rather than failing immediately or requiring explicit readiness checks from callers.

Shared Processor Pattern in Koin Wiring

The dependency injection module in appModule.kt (iOS variant, lines 42-45) registers the same ElevenLabProcessor instance for all four provider interfaces:

single<TTSProvider> { ElevenLabProcessor(get(), get()) }
single<AudioIsolationProvider> { ElevenLabProcessor(get(), get()) }
single<SoundEffectProvider> { ElevenLabProcessor(get(), get()) }
single<STTProvider> { ElevenLabProcessor(get(), get()) }

Because Koin scopes these as singletons, every provider shares the same Flow subscription and client instance. When the apiKey flow emits a new value, all providers simultaneously receive the updated client without individual reinitialization logic.

Practical Implementation Examples

Observing API Key State in Compose

UI layers collect the apiKey flow to conditionally render authentication prompts:

val apiKey by accountManager.apiKey.collectAsState(initial = null)
if (apiKey == null) {
    Text("Please enter your ElevenLabs API key")
}

Reacting to Subscription Changes

Monitor character usage limits by collecting the subscription status flow:

val subscription by accountManager.subscriptionStatus
    .collectAsState(initial = null)

subscription?.let {
    Text("You have ${it.characterCount} / ${it.characterLimit} characters used")
}

Forcing Initialization Manually

While rarely necessary, you can trigger initialization explicitly by casting the provider to its implementation class:

val processor: TTSProvider = get<TTSProvider>()
(processor as? ElevenLabProcessor)?.initialize()

Summary

• AccountManager exposes apiKey and subscriptionStatus as reactive Flow instances backed by DataStore.
• ElevenLabProcessor subscribes to the API key flow using distinctUntilChanged and collectLatest to lazily create and recreate the ElevenLabs client only when credentials change.
• requireClient() implements a retry mechanism with 200ms delays to gracefully handle calls made before flow collection completes.
• Koin singleton scoping ensures all four provider interfaces share the same flow subscription and client state.

Frequently Asked Questions

What happens when the API key changes at runtime?

When the user updates their API key through the settings screen, the DataStore write triggers the apiKey flow to emit a new value. The ElevenLabProcessor detects this change via distinctUntilChanged and uses collectLatest to cancel the existing client and initialize a new ElevenLabsClient instance with the updated credentials.

How does the provider handle calls made before initialization completes?

The requireClient() function checks if the client is initialized and, if not, triggers initialize() before waiting 200 milliseconds and retrying up to two times. This ensures that provider methods like generate() or transcribeAudio() block briefly for the flow-driven initialization rather than throwing immediate exceptions.

Why use Flow instead of suspend functions for credential access?

Flow provides continuous, reactive updates that enable automatic client reconfiguration without polling or manual refresh calls. Unlike one-shot suspend functions, the cold stream pattern allows providers to remain dormant until valid credentials exist and automatically adapt to configuration changes throughout the application lifecycle.

Can multiple provider types share the same API key state?

Yes. The Koin module registers ElevenLabProcessor as a singleton for all four provider interfaces (TTSProvider, STTProvider, etc.), meaning all instances share the same AccountManager reference and apiKey flow subscription. Updating the key once immediately reconfigures every provider simultaneously.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}