package com.fabledsword.minstrel.player

import android.net.Uri
import androidx.media3.datasource.DataSpec
import androidx.media3.datasource.cache.CacheDataSource
import androidx.media3.datasource.cache.CacheWriter
import androidx.media3.datasource.okhttp.OkHttpDataSource
import com.fabledsword.minstrel.auth.AuthStore
import com.fabledsword.minstrel.di.ApplicationScope
import com.fabledsword.minstrel.shared.resolveServerUrl
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import okhttp3.OkHttpClient
import javax.inject.Inject
import javax.inject.Singleton

/**
 * Pre-downloads the next-N tracks in the queue into the shared Media3
 * [androidx.media3.datasource.cache.SimpleCache] so a skip-forward or
 * natural advance plays from disk instead of waiting on a fresh HTTP
 * connection. Mirrors the Flutter `Prefetcher` (cache/prefetcher.dart):
 * watches the player's current track, walks forward by
 * [com.fabledsword.minstrel.cache.audiocache.CacheSettings.prefetchWindow]
 * tracks, and pins each one. Idempotent — `CacheWriter` is a no-op when
 * a track is already fully resident, so the reconcile is cheap on
 * every queue/index change.
 *
 * Eager-constructed via the construct-the-singleton trick in
 * [com.fabledsword.minstrel.MinstrelApplication]; the init block wires
 * the observer.
 */
@Singleton
class AudioPrefetcher @Inject constructor(
    @ApplicationScope private val scope: CoroutineScope,
    private val playerController: PlayerController,
    private val playerFactory: PlayerFactory,
    private val authStore: AuthStore,
    private val okHttpClient: OkHttpClient,
) {
    // Reuse the same cache + upstream chain the player uses. Without the
    // SimpleCache reference here the prefetcher would write to a
    // disjoint store and playback would never read what we pinned.
    private val cacheDataSourceFactory: CacheDataSource.Factory =
        CacheDataSource.Factory()
            .setCache(playerFactory.simpleCache)
            .setUpstreamDataSourceFactory(OkHttpDataSource.Factory(okHttpClient))

    // trackId -> in-flight job. Reconcile cancels jobs whose track has
    // dropped out of the window (skip-prev, queue rebuild) so a stale
    // prefetch doesn't keep the network busy.
    private val activeJobs = mutableMapOf<String, Job>()
    private val mutex = Mutex()

    private data class ReconcileInput(
        val queue: List<Pair<String, String>>,
        val index: Int,
        val window: Int,
        val isPlaying: Boolean,
    )

    init {
        scope.launch {
            combine(
                playerController.uiState.map { it.queue.map { t -> t.id to t.streamUrl } },
                playerController.uiState.map { it.queueIndex },
                authStore.cacheSettings.map { it.prefetchWindow },
                playerController.uiState.map { it.isPlaying },
            ) { queue, index, window, isPlaying ->
                ReconcileInput(queue, index, window, isPlaying)
            }
                .distinctUntilChanged()
                .collect { input ->
                    reconcile(input.queue, input.index, input.window, input.isPlaying)
                }
        }
    }

    /**
     * Apply the prefetch window to the current queue.
     *
     * Cancellation of out-of-window jobs always runs -- a queue mutation
     * or skip should free bandwidth from stale prefetches immediately.
     * Starting new prefetches is gated on [isPlaying]: until the current
     * track is actually playing, every byte of upstream bandwidth should
     * land on it, not on upcoming-track prefetches. Without this gate a
     * cold start fanned out 4-6 concurrent CacheWriter jobs against the
     * same OkHttp client as the playback DataSource and the user waited
     * ~25 s for the first audio to start; with the gate the current
     * track gets the full pipe to its first STATE_READY, then the
     * prefetcher fills in the next window.
     */
    private suspend fun reconcile(
        queue: List<Pair<String, String>>,
        index: Int,
        window: Int,
        isPlaying: Boolean,
    ) {
        mutex.withLock {
            val targets = computeTargets(queue, index, window)
            if (targets.isEmpty()) {
                cancelAllLocked()
                return
            }
            val targetIds = targets.mapTo(mutableSetOf()) { it.first }
            // Cancellation always runs so a queue mutation or skip frees
            // the pipe immediately, even while paused.
            cancelOutOfWindowLocked(targetIds)
            if (isPlaying) startInWindowLocked(targets)
        }
    }

    private fun computeTargets(
        queue: List<Pair<String, String>>,
        index: Int,
        window: Int,
    ): List<Pair<String, String>> {
        // Exclude the currently-playing track (it's loaded by the player
        // itself) and walk `window` tracks forward.
        val firstIdx = index + 1
        val lastIdx = (index + window).coerceAtMost(queue.size - 1)
        val isValid = index >= 0 && queue.isNotEmpty() && window > 0 && firstIdx <= lastIdx
        return if (isValid) queue.subList(firstIdx, lastIdx + 1) else emptyList()
    }

    private fun cancelOutOfWindowLocked(targetIds: Set<String>) {
        activeJobs.entries
            .filter { it.key !in targetIds }
            .toList()
            .forEach { (id, job) ->
                job.cancel()
                activeJobs.remove(id)
            }
    }

    private fun startInWindowLocked(targets: List<Pair<String, String>>) {
        // Skip blank URLs (these come from minimal TrackRefs synthesized
        // from playlist rows when the upstream track was removed from
        // the library).
        for ((trackId, streamUrl) in targets) {
            if (trackId in activeJobs || streamUrl.isBlank()) continue
            val job = scope.launch(Dispatchers.IO) {
                runCatching { prefetchOne(trackId, streamUrl) }
                mutex.withLock { activeJobs.remove(trackId) }
            }
            activeJobs[trackId] = job
        }
    }

    private fun cancelAllLocked() {
        activeJobs.values.forEach { it.cancel() }
        activeJobs.clear()
    }

    private fun prefetchOne(trackId: String, streamUrl: String) {
        // Match PlayerController.toMediaItem's resolveServerUrl +
        // setCustomCacheKey(id) so the cached bytes are keyed exactly
        // like a normal playback request — otherwise the player would
        // miss them on read-through.
        val resolved = resolveServerUrl(streamUrl) ?: streamUrl
        val dataSpec = DataSpec.Builder()
            .setUri(Uri.parse(resolved))
            .setKey(trackId)
            .build()
        val writer = CacheWriter(
            cacheDataSourceFactory.createDataSource(),
            dataSpec,
            /* temporaryBuffer = */ null,
            /* progressListener = */ null,
        )
        // Blocking, but we're on Dispatchers.IO. Throws on network
        // failure — outer runCatching swallows so a single failed
        // prefetch doesn't trip the whole reconcile.
        writer.cache()
    }
}
