package com.ponzischeme89.memby.data

import androidx.lifecycle.Lifecycle
import androidx.lifecycle.ProcessLifecycleOwner
import androidx.lifecycle.repeatOnLifecycle
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.StateFlow
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

/**
 * Keeps this television's settings and the viewer's server-held document the same thing.
 *
 * Three events move settings, and all three go through here:
 *
 * - **The viewer changes something on this TV.** The local write already happened —
 *   nothing in the UI waits on the network — and the new document is pushed up after it.
 * - **The viewer changes something on another TV, or an operator pushes from the admin
 *   console.** The revision on the status poll goes up, this notices and pulls.
 * - **A profile signs in somewhere for the first time.** The server's copy wins if it has
 *   one; otherwise this TV's settings become the starting document.
 *
 * The failure mode being designed against is a *loop*: a pull that looks like a local
 * change and triggers a push, which bumps the revision, which looks like a remote change.
 * [lastSynced] is what breaks it — it records the exact document both ends agreed on, and
 * a push happens only when the local one differs from it. Adopting a pull sets it, so the
 * write that follows never looks like news.
 *
 * Everything here fails silently and retries on the next event. Settings are not worth an
 * error message on a television, and the state they are in — local values intact, server
 * copy intact — is a correct one to be left in until the next poll.
 */
class PreferencesSync(
    private val repository: EmbyRepository,
    private val settings: SettingsStore,
    private val remoteRevision: StateFlow<Long>,
    private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.IO),
) {
    /**
     * The document last known to be identical on both ends, per profile. A local state
     * matching this is not a change and must not be pushed.
     */
    private var lastSynced: Pair<String, UserPreferences>? = null

    /** One sync at a time: a pull and a push racing would decide the revision twice. */
    private val mutex = Mutex()

    init {
        scope.launch { run() }
    }

    private suspend fun run() {
        // Nothing to sync for a TV nobody is looking at, and the status poll this rides
        // beside stops in the background too, so the revision would go stale anyway.
        ProcessLifecycleOwner.get().lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
            combine(
                repository.settingsFlow,
                remoteRevision,
            ) { session, revision -> session to revision }
                // The launcher rewrites the home cache constantly and it is part of
                // Settings; without this, every refresh would look like a settings change
                // and be considered for a push.
                .map { (session, revision) -> SyncTrigger(session, revision) }
                .distinctUntilChanged()
                // Plain collect, not collectLatest: a reconcile that is cancelled halfway
                // could leave [lastSynced] describing a document that was never written.
                // These are short and serialised by the mutex, so waiting is cheaper than
                // reasoning about a half-applied sync.
                .collect(::reconcile)
        }
    }

    /** The only parts of the session a sync depends on. */
    private data class SyncTrigger(
        val profileKey: String,
        val signedIn: Boolean,
        val localRevision: Long,
        val remoteRevision: Long,
        val local: UserPreferences,
    ) {
        constructor(session: Settings, remoteRevision: Long) : this(
            profileKey = "${session.userId.orEmpty()}@${session.serverUrl.orEmpty()}",
            signedIn = session.isSignedIn,
            localRevision = session.preferencesRevision,
            remoteRevision = remoteRevision,
            local = session.toUserPreferences(),
        )
    }

    private suspend fun reconcile(trigger: SyncTrigger) {
        if (!ServerConfig.isGateway || !trigger.signedIn) return
        mutex.withLock {
            val agreed = lastSynced?.takeIf { it.first == trigger.profileKey }?.second

            // Never synced on this TV, or the server has moved on without us. Pull first:
            // the server's copy is the shared truth, and a push here would overwrite a
            // change made on another television with this one's defaults.
            if (agreed == null || trigger.remoteRevision > trigger.localRevision) {
                if (pull(trigger)) return
            }

            // Only a genuine local edit gets pushed. Anything else is either the document
            // just adopted or an unrelated part of Settings changing.
            val current = lastSynced?.takeIf { it.first == trigger.profileKey }?.second
            if (current != null && current != trigger.local) push(trigger)
        }
    }

    /**
     * Takes the server's copy. Returns true when the local document was replaced, so the
     * caller knows not to treat the write it just made as a local edit.
     */
    private suspend fun pull(trigger: SyncTrigger): Boolean {
        val remote = runCatching { repository.userPreferences() }.getOrNull() ?: return false
        // Revision 0 means the server has never been told anything about this viewer. The
        // settings already on this television are then the best starting point there is,
        // so they are pushed up rather than replaced by catalogue defaults.
        if (remote.revision == 0L) {
            lastSynced = trigger.profileKey to UserPreferences()
            push(trigger)
            return true
        }
        val decoded = decodeUserPreferences(remote.preferences, fallback = trigger.local)
        lastSynced = trigger.profileKey to decoded
        // Written even when the values match, because the revision must advance: without
        // it the next poll sees the same gap and pulls again, every ten seconds, forever.
        runCatching { settings.applyRemotePreferences(decoded, remote.revision) }
            .onFailure { lastSynced = null }
        return true
    }

    private suspend fun push(trigger: SyncTrigger) {
        val stored = runCatching {
            repository.saveUserPreferences(trigger.localRevision, trigger.local.encode())
        }.getOrNull() ?: return

        // A 409 comes back here as an ordinary result carrying somebody else's document —
        // an operator's push, nearly always. Adopting it is the correct outcome: this TV
        // was editing a revision that no longer exists, and retrying would revert them.
        val decoded = decodeUserPreferences(stored.preferences, fallback = trigger.local)
        lastSynced = trigger.profileKey to decoded
        if (decoded == trigger.local) {
            runCatching { settings.setPreferencesRevision(stored.revision) }
        } else {
            // The request may have been in flight while the viewer made a newer choice.
            // A conflict describes the document that won against the request, not against
            // an edit made after it. Advance that newer edit to the winning revision and
            // let the queued trigger push it, instead of replacing it with stale values.
            val current = runCatching { settings.snapshot().toUserPreferences() }.getOrNull()
            if (current != null && current != trigger.local) {
                runCatching { settings.setPreferencesRevision(stored.revision) }
            } else {
                runCatching { settings.applyRemotePreferences(decoded, stored.revision) }
                    .onFailure { lastSynced = null }
            }
        }
    }
}
