package link.roadio.riff.userstate.android

import android.app.Application
import android.content.Context
import android.os.Bundle
import android.os.ParcelFileDescriptor
import android.os.SystemClock
import android.util.Log
import link.roadio.riff.storage.hash.nativeXxh3
import link.roadio.riff.storage.root.ApplicationDataStorage
import link.roadio.riff.userstate.ContentIds
import link.roadio.riff.userstate.UserSnapshotStore
import link.roadio.riff.userstate.UserStateCoordinator
import link.roadio.riff.userstate.model.UserStateCodec
import java.io.File
import java.util.concurrent.CompletableFuture
import java.util.concurrent.Executors
import java.util.concurrent.ScheduledFuture
import java.util.concurrent.TimeUnit

/** The provider and service share this one Indexer-process writer, independently of scan work. */
object IndexerUserState {
    private val executor = Executors.newSingleThreadScheduledExecutor()
    private var live: Live? = null
    private var timer: ScheduledFuture<*>? = null
    private var publicationFailed = false

    @Volatile private var problemListener: ((Boolean) -> Unit)? = null

    fun observePublicationProblem(listener: ((Boolean) -> Unit)?) =
        task {
            problemListener = listener
            listener?.invoke(publicationFailed)
        }

    @Volatile var onSaved: ((List<String>) -> Unit)? = null

    private data class Live(
        val context: Context,
        val path: File,
        val rootId: Long,
        val store: UserSnapshotStore,
        val coordinator: UserStateCoordinator,
        val ingress: link.roadio.riff.userstate.UserStateIngress =
            link.roadio.riff.userstate
                .UserStateIngress(),
        var fenced: Boolean = false,
    )

    fun open(
        context: Context,
        writer: Long? = null,
        confirmed: Map<Pair<Int, Int>, Long> = emptyMap(),
        expectedRoot: Long? = null,
    ): Bundle =
        task {
            val state = ensure(context)
            check(!state.fenced && (expectedRoot == null || expectedRoot == state.rootId)) { "Closed user-state root admission" }
            require(confirmed.isEmpty() || expectedRoot != null)
            writer?.let {
                publication(state) { state.coordinator.openWriter(it) }
                val headers =
                    mapOf((0 to 0) to state.coordinator.installation().header) +
                        (1..3).associate { slot -> (1 to slot) to state.coordinator.profile(slot).header }
                state.ingress.open(it, headers, confirmed)
            }
            state.store.openCurrent { ParcelFileDescriptor.open(it, ParcelFileDescriptor.MODE_READ_ONLY) }.let { descriptors ->
                Bundle().apply {
                    putInt("ownerPid", android.os.Process.myPid())
                    putLong("root", state.rootId)
                    descriptors.forEach { (key, descriptor) ->
                        putParcelable("${key.kind}:${key.slot}", descriptor)
                    }
                }
            }
        }.get(35, TimeUnit.SECONDS)

    fun submit(
        context: Context,
        fields: List<String>,
        payload: ParcelFileDescriptor,
    ): List<String> =
        task {
            payload.use {
                require(fields.size == 1)
                val bytes =
                    ParcelFileDescriptor.AutoCloseInputStream(ParcelFileDescriptor.dup(payload.fileDescriptor)).use { input ->
                        val out = java.io.ByteArrayOutputStream()
                        val block = ByteArray(8192)
                        while (true) {
                            val count = input.read(block)
                            if (count < 0) break
                            require(count > 0 && out.size() <= MAX_BYTES - count)
                            out.write(block, 0, count)
                        }
                        out.toByteArray()
                    }
                val fingerprint = nativeXxh3(bytes)
                require(fingerprint == fields.single().toLong())
                val command = UserStateCodec.decodeUserStateCommand(bytes)
                require(command.protocolVersion == 1 && command.writes.size in 1..4)
                val writer = checkNotNull(command.writerId)
                val state = ensure(context)
                check(!state.fenced && checkNotNull(command.rootId) == state.rootId) { "Closed user-state root" }
                val scopes = command.writes.map { it.kind to it.profileSlot }
                require(scopes.distinct().size == scopes.size)
                command.writes.forEach { write ->
                    require(write.sequence > 0)
                    when (write.kind) {
                        0 -> {
                            require(write.profileSlot == 0 && write.profile == null && write.start == null)
                            val incoming = checkNotNull(write.installation)
                            link.roadio.riff.userstate.UserStateValidation
                                .installation(incoming)
                        }

                        1 -> {
                            require(write.profileSlot in 1..3 && write.installation == null)
                            val incoming = checkNotNull(write.profile)
                            link.roadio.riff.userstate.UserStateValidation
                                .header(incoming.header, write.profileSlot)
                            AndroidUserStateValidation.profile(incoming)
                        }

                        else -> {
                            error("Unsupported user-state scope")
                        }
                    }
                    state.ingress.check(writer, write.kind, write.profileSlot, write.sequence, fingerprint)
                }
                val results =
                    state.coordinator.atomic {
                        command.writes.map { write ->
                            when {
                                write.kind == 0 -> {
                                    changeInstallation(writer, write.sequence) { checkNotNull(write.installation) }
                                }

                                write.start != null -> {
                                    start(
                                        writer,
                                        write.sequence,
                                        write.profileSlot,
                                        checkNotNull(write.start),
                                    ) { checkNotNull(write.profile) }
                                }

                                else -> {
                                    changeProfile(writer, write.sequence, write.profileSlot) { checkNotNull(write.profile) }
                                }
                            }
                        }
                    }
                command.writes.forEach { state.ingress.applied(it.kind, it.profileSlot, it.sequence, fingerprint) }
                schedule()
                listOf("USER_STATE") +
                    command.writes.flatMapIndexed { index, write ->
                        val generation =
                            if (write.kind ==
                                0
                            ) {
                                state.coordinator
                                    .installation()
                                    .header.generation
                            } else {
                                state.coordinator
                                    .profile(write.profileSlot)
                                    .header.generation
                            }
                        val history =
                            if (write.start != null) {
                                state.coordinator
                                    .history(write.profileSlot)
                                    .header.generation
                            } else {
                                0L
                            }
                        listOf(
                            write.kind.toString(),
                            write.profileSlot.toString(),
                            write.sequence.toString(),
                            results[index].name,
                            generation.toString(),
                            history.toString(),
                        )
                    }
            }
        }.get(35, TimeUnit.SECONDS)

    data class ResetResult(
        val accepted: Boolean,
        val completed: Boolean,
    )

    /** Fence the writer before retiring Internal paths; unlike relocation this explicitly discards RAM. */
    fun resetToInternal(
        context: Context,
        expected: link.roadio.riff.storage.root.ApplicationDataSelection,
    ): ResetResult =
        task {
            val storage = ApplicationDataStorage(context)
            if (storage.selection() != expected || storage.selectedRootAvailable()) return@task ResetResult(false, false)
            val prior = live
            val wasFenced = prior?.fenced ?: false
            timer?.cancel(false)
            timer = null
            prior?.fenced = true
            try {
                if (!storage.resetToInternal(expected)) {
                    prior?.fenced = wasFenced
                    if (!wasFenced) schedule()
                    ResetResult(false, false)
                } else {
                    live = null
                    setPublicationProblem(false)
                    ResetResult(true, true)
                }
            } catch (error: Exception) {
                // Bootstrap recovery finishes an interrupted confirmed reset at the next admission.
                // Never let an old in-memory writer populate the fresh path after a partial reset.
                live = null
                Log.w("RiffUserState", "confirmed_internal_reset_incomplete", error)
                ResetResult(true, false)
            }
        }.get(35, TimeUnit.SECONDS)

    fun fence(): Boolean =
        task {
            live?.let { state ->
                if (!state.fenced) {
                    publication(state) { state.coordinator.flush() }
                    state.fenced = true
                }
            }
            timer?.cancel(false)
            timer = null
            true
        }.get(35, TimeUnit.SECONDS)

    fun activate() =
        task {
            live = null
            timer?.cancel(false)
            timer = null
        }.get(35, TimeUnit.SECONDS)

    fun drainAsync() =
        task {
            live?.let { state -> publication(state) { state.coordinator.flush() } }
            true
        }

    fun drain(): Boolean = drainAsync().get(35, TimeUnit.SECONDS)

    fun prepare(context: Context) =
        task {
            ensure(context)
            true
        }

    private fun ensure(context: Context): Live {
        check(Application.getProcessName().endsWith(":indexer"))
        val storage = ApplicationDataStorage(context)
        val root = checkNotNull(storage.activeRootOrNull()) { "Selected storage is unavailable" }
        live?.let {
            check(it.path == root) { "User-state root changed without fence" }
            return it
        }
        val directory = File(root, "indexer/user-state")
        if (!File(directory, "CURRENT").exists() && !File(directory, "PREVIOUS").exists()) {
            check(
                File(root, "player").listFiles().orEmpty().none {
                    it.name.endsWith(".sqlite") ||
                        it.name.startsWith("player-state.sqlite-")
                },
            ) {
                "Incompatible Player state requires explicit storage reset; no SQLite importer is provided"
            }
        }
        val ids = ContentIds()
        val rootId =
            UserSnapshotStore.existingRootId(directory)
                ?: RuntimeIds.reference("user-store", listOf(context.packageName, root.absolutePath)).toULong(16).toLong()
        val store = UserSnapshotStore(directory, rootId, profileValidation = AndroidUserStateValidation::profile)
        val coordinator =
            UserStateCoordinator(
                store,
                ids,
                SystemClock::elapsedRealtime,
                bookmarkMinimumUs =
                    Math.multiplyExact(
                        link.roadio.riff.shared.config.AndroidProjectConfiguration
                            .load(
                                context,
                            ).snapshot
                            .number("player.bookmarkMinSpacingSeconds")
                            .toLong(),
                        1_000_000L,
                    ),
                historyDecay =
                    link.roadio.riff.shared.config.AndroidProjectConfiguration
                        .load(
                            context,
                        ).snapshot
                        .number("ai.selectionModel.historyDecay"),
            ) { receipts ->
                receipts.forEach { receipt ->
                    onSaved?.invoke(
                        listOf(
                            rootId.toString(),
                            receipt.kind.toString(),
                            receipt.slot.toString(),
                            receipt.writer.toString(),
                            receipt.sequence.toString(),
                            receipt.generation.toString(),
                        ),
                    )
                }
            }
        return Live(context.applicationContext, root, rootId, store, coordinator).also { live = it }
    }

    private fun schedule() {
        timer?.cancel(false)
        val state = live ?: return
        val deadline = state.coordinator.deadline() ?: return
        timer =
            executor.schedule({
                runCatching { publication(state) { state.coordinator.flushIfDue() } }
                    .onFailure { Log.w("RiffUserState", "user_state_publication_failed", it) }
                // On failure retain the pending cut and apply a bounded retry instead of a tight loop.
                if (state.coordinator.deadline() != null) {
                    timer = executor.schedule({ schedule() }, 1000, TimeUnit.MILLISECONDS)
                }
            }, (deadline - SystemClock.elapsedRealtime()).coerceAtLeast(0), TimeUnit.MILLISECONDS)
    }

    private fun <T> publication(
        state: Live,
        action: () -> T,
    ): T =
        try {
            check(!state.fenced && ApplicationDataStorage(state.context).activeRootOrNull() == state.path) {
                "Selected user-state storage is unavailable"
            }
            action().also { setPublicationProblem(false) }
        } catch (error: Exception) {
            setPublicationProblem(true)
            throw error
        }

    private fun setPublicationProblem(failed: Boolean) {
        if (publicationFailed == failed) return
        publicationFailed = failed
        problemListener?.invoke(failed)
    }

    private fun <T> task(action: () -> T): CompletableFuture<T> = CompletableFuture.supplyAsync(action, executor)

    private const val MAX_BYTES = 64 * 1024 * 1024
}