diff --git a/android/app/src/main/java/com/fabledsword/minstrel/MinstrelApplication.kt b/android/app/src/main/java/com/fabledsword/minstrel/MinstrelApplication.kt index 8084b845..e8f362f2 100644 --- a/android/app/src/main/java/com/fabledsword/minstrel/MinstrelApplication.kt +++ b/android/app/src/main/java/com/fabledsword/minstrel/MinstrelApplication.kt @@ -9,6 +9,7 @@ import coil3.network.okhttp.OkHttpNetworkFetcherFactory import com.fabledsword.minstrel.cache.mutations.MutationReplayer import com.fabledsword.minstrel.cache.sync.SyncController import com.fabledsword.minstrel.di.ApplicationScope +import com.fabledsword.minstrel.events.EventsStream import com.fabledsword.minstrel.player.PlayEventsReporter import com.fabledsword.minstrel.player.ResumeController import dagger.hilt.android.HiltAndroidApp @@ -68,6 +69,15 @@ class MinstrelApplication : */ @Suppress("unused") @Inject lateinit var playEventsReporter: PlayEventsReporter + /** + * Same construct-the-singleton trick — EventsStream's init block + * subscribes to AuthStore.sessionCookie and opens / closes the + * `/api/events/stream` SSE subscription on auth transitions. + * Without this @Inject the events SharedFlow would have no + * upstream and consumers would never see anything. + */ + @Suppress("unused") @Inject lateinit var eventsStream: EventsStream + @Inject @ApplicationScope lateinit var appScope: CoroutineScope override fun onCreate() { diff --git a/android/app/src/main/java/com/fabledsword/minstrel/events/EventsStream.kt b/android/app/src/main/java/com/fabledsword/minstrel/events/EventsStream.kt new file mode 100644 index 00000000..5ed0fc90 --- /dev/null +++ b/android/app/src/main/java/com/fabledsword/minstrel/events/EventsStream.kt @@ -0,0 +1,141 @@ +package com.fabledsword.minstrel.events + +import com.fabledsword.minstrel.auth.AuthStore +import com.fabledsword.minstrel.di.ApplicationScope +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.launch +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.contentOrNull +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import okhttp3.sse.EventSource +import okhttp3.sse.EventSourceListener +import okhttp3.sse.EventSources +import javax.inject.Inject +import javax.inject.Singleton + +private const val SSE_PATH = "/api/events/stream" +private const val EVENTS_BUFFER_CAPACITY = 64 + +/** + * Long-lived SSE subscription to `GET /api/events/stream`. Exposes + * [events] as a process-wide [SharedFlow]; consumers (per-screen + * ViewModels + the central [LiveEventsDispatcher]) collect filtered + * subsets of the stream. + * + * Connection lifecycle mirrors + * `flutter_client/lib/shared/live_events_provider.dart`: + * - Gated on having a session cookie. Subscription opens when the + * cookie transitions to non-null and closes when it transitions + * back to null (sign-out). + * - No client-side timeout — the server emits 15s heartbeats which + * okhttp-sse handles transparently. + * - No explicit reconnect-with-backoff in v1; if the stream drops + * mid-session (server restart, network blip) it stays closed until + * the next auth-cookie transition. Matches Flutter's behavior. + * + * The URL passes through the placeholder host that + * `BaseUrlInterceptor` rewrites — same mechanism the rest of the + * Retrofit traffic uses, so server-URL changes propagate without + * needing a separate path through here. + */ +@Singleton +class EventsStream @Inject constructor( + private val authStore: AuthStore, + @ApplicationScope private val scope: CoroutineScope, + private val okHttpClient: OkHttpClient, + private val json: Json, +) { + private val factory = EventSources.createFactory(okHttpClient) + + private val emitter = MutableSharedFlow( + replay = 0, + extraBufferCapacity = EVENTS_BUFFER_CAPACITY, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) + val events: SharedFlow = emitter.asSharedFlow() + + private var currentSource: EventSource? = null + + init { + scope.launch { + authStore.sessionCookie + .map { !it.isNullOrEmpty() } + .distinctUntilChanged() + .collect { signedIn -> + currentSource?.cancel() + currentSource = if (signedIn) openSubscription() else null + } + } + } + + private fun openSubscription(): EventSource { + val request = Request.Builder() + .url("http://placeholder.invalid$SSE_PATH") + .header("Accept", "text/event-stream") + .build() + return factory.newEventSource(request, Listener()) + } + + private inner class Listener : EventSourceListener() { + override fun onEvent( + eventSource: EventSource, + id: String?, + type: String?, + data: String, + ) { + parseFrame(kind = type, data = data)?.let(emitter::tryEmit) + } + + override fun onClosed(eventSource: EventSource) { + // Stream ended (server-side close or network drop). No + // explicit reconnect — auth-cookie transitions re-open. + } + + override fun onFailure( + eventSource: EventSource, + t: Throwable?, + response: Response?, + ) { + // Same as onClosed — auth transition will re-open. The + // throwable/response carry diagnostics; we don't surface + // them in v1 since the stream is "best effort" by design. + } + } + + /** + * Parses one SSE frame into a [LiveEvent]. The SSE `event:` field + * carries the kind; the `data:` field is a JSON object with + * `kind`/`user_id`/`data` (the kind in the payload is redundant + * with the SSE event field — we prefer the SSE field, falling + * back to the payload field). Heartbeat comment frames don't + * reach here (okhttp-sse filters them). + */ + private fun parseFrame(kind: String?, data: String): LiveEvent? = try { + val payload = json.parseToJsonElement(data).jsonObject + val resolvedKind = kind ?: payload["kind"]?.jsonPrimitive?.contentOrNull + if (resolvedKind.isNullOrEmpty()) { + null + } else { + LiveEvent( + kind = resolvedKind, + userId = payload["user_id"]?.jsonPrimitive?.contentOrNull.orEmpty(), + data = (payload["data"] as? JsonObject) ?: JsonObject(emptyMap()), + ) + } + } catch ( + @Suppress("TooGenericExceptionCaught") _: Throwable, + ) { + null + } +} diff --git a/android/app/src/main/java/com/fabledsword/minstrel/events/LiveEvent.kt b/android/app/src/main/java/com/fabledsword/minstrel/events/LiveEvent.kt new file mode 100644 index 00000000..c296b25f --- /dev/null +++ b/android/app/src/main/java/com/fabledsword/minstrel/events/LiveEvent.kt @@ -0,0 +1,21 @@ +package com.fabledsword.minstrel.events + +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.JsonObject + +/** + * Parsed event from the server's SSE stream. Mirrors + * `flutter_client/lib/shared/live_events_provider.dart`'s `LiveEvent`. + * + * - [kind] is the SSE `event:` field (e.g. "track.liked", "playlist.deleted"). + * - [userId] is the actor whose user-scoped state changed (empty for + * broadcast events like scan.* that aren't user-bound). + * - [data] is the raw event payload; consumers deserialize into the + * typed shape they expect for the kind they handle. + */ +@Serializable +data class LiveEvent( + val kind: String, + val userId: String, + val data: JsonObject, +)