package link.roadio.riff.player.storage

import android.content.Context
import android.net.Uri
import android.os.Bundle
import android.os.ParcelFileDescriptor
import android.os.Process
import android.util.Log
import link.roadio.riff.player.library.PlayerLibraryRepository
import link.roadio.riff.storage.hash.nativeXxh3
import link.roadio.riff.userstate.ContentIds
import link.roadio.riff.userstate.UserStateReplica
import link.roadio.riff.userstate.model.InstallationStateModel
import link.roadio.riff.userstate.model.ListeningHistoryModel
import link.roadio.riff.userstate.model.ProfileStateModel
import link.roadio.riff.userstate.model.UserStateCodec
import link.roadio.riff.userstate.model.UserStateCommandModel
import link.roadio.riff.userstate.model.UserStateWriteModel
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicBoolean

/** No writable Player files. Frozen outbox payloads travel on the library's existing channel. */
internal class PlayerStateSession private constructor(
    private val context: Context,
    val replica: UserStateReplica,
) : AutoCloseable {
    private val repository =
        PlayerLibraryRepository.get(
            context,
            (
                checkNotNull(
                    link.roadio.riff.shared.config.AndroidProjectConfiguration
                        .currentOrNull(),
                ).snapshot.value("player.excludedGenreIds") as List<*>
            ).map {
                (it as String).toLong()
            }.toSet(),
        )
    private val executor = Executors.newSingleThreadScheduledExecutor()
    private val pipeWriter = Executors.newSingleThreadExecutor()
    private val scheduled = AtomicBoolean()
    private var staged: Pair<UserStateReplica.Mutation, ByteArray>? = null

    @Volatile private var closed = false
    private val savedListener: (List<String>) -> Unit = { fields ->
        // This callback must never wait for publication, Provider or another reliable command.
        replica.saved(fields[0].toLong(), fields[1].toInt(), fields[2].toInt(), fields[3].toLong(), fields[4].toLong(), fields[5].toLong())
        wake()
    }

    init {
        repository.attachUserState(savedListener) {
            executor.execute {
                if (!closed) {
                    runCatching {
                        reconcile(
                            snapshot(context, replica.writerId, confirmed = replica.confirmed(), expectedRoot = replica.rootId),
                            replica,
                        )
                        replica.replay()
                        wake()
                    }.onFailure {
                        Log.w(TAG, "user_state_reconnect_failed", it)
                        wake()
                    }
                }
            }
        }
    }

    fun wake() {
        if (!closed && scheduled.compareAndSet(false, true)) executor.execute(::pump)
    }

    private fun pump() {
        var retry = false
        try {
            while (!closed) {
                val mutation = replica.next() ?: break
                val members = mutation.group
                val bytes =
                    staged?.takeIf { it.first === mutation }?.second ?: UserStateCodec
                        .encode(
                            UserStateCommandModel(
                                protocolVersion = 1,
                                rootId = replica.rootId,
                                writerId = replica.writerId,
                                writes =
                                    members.map {
                                        UserStateWriteModel(
                                            it.kind,
                                            it.slot,
                                            it.sequence,
                                            it.profile,
                                            it.installation,
                                            it.start,
                                        )
                                    },
                            ),
                        ).also { staged = mutation to it }
                require(bytes.size <= 64 * 1024 * 1024) { "User-state command exceeds descriptor bound" }
                val fields = listOf(nativeXxh3(bytes).toString())
                val pipe = ParcelFileDescriptor.createReliablePipe()
                val writing =
                    java.util.concurrent.CompletableFuture.runAsync({
                        ParcelFileDescriptor.AutoCloseOutputStream(pipe[1]).use { it.write(bytes) }
                    }, pipeWriter)
                try {
                    val result = pipe[0].use { repository.sendUserState(fields, it) }
                    writing.get(35, TimeUnit.SECONDS)
                    require(result.size == 1 + 6 * members.size && result.first() == "USER_STATE")
                    result.drop(1).chunked(6).forEachIndexed { index, row ->
                        val member = members[index]
                        require(row[0].toInt() == member.kind && row[1].toInt() == member.slot && row[2].toLong() == member.sequence)
                        if (row[3] == "SAVED") {
                            replica.saved(replica.rootId, member.kind, member.slot, replica.writerId, member.sequence, row[4].toLong())
                            if (member.start !=
                                null
                            ) {
                                replica.saved(replica.rootId, 2, member.slot, replica.writerId, member.sequence, row[5].toLong())
                            }
                        } else {
                            require(row[3] == "ACCEPTED")
                            replica.accepted(member)
                        }
                    }
                    // Immutable semantic models remain until SAVED; deterministic IPC bytes can be
                    // regenerated after reconnection. Keep at most one large encoded payload in RAM.
                    staged = null
                } finally {
                    runCatching { pipe[0].close() }
                    runCatching { pipe[1].close() }
                }
            }
        } catch (error: Exception) {
            Log.w(TAG, "user_state_delivery_pending", error)
            retry = true
        } finally {
            scheduled.set(false)
            if (!closed && (retry || replica.next() != null)) executor.schedule(::wake, if (retry) 1000 else 0, TimeUnit.MILLISECONDS)
        }
    }

    /** Completion is awaited by existing background commit paths, never by Binder/ACK dispatch. */
    fun awaitSaved() {
        wake()
        replica.completion().get()
    }

    fun discardForConfirmedReset() {
        closed = true
        replica.discardForConfirmedReset()
        repository.detachUserState(savedListener)
        executor.shutdownNow()
        pipeWriter.shutdownNow()
        staged = null
    }

    override fun close() {
        wake()
        // Relocation is an explicit drain boundary. Normal mutations use the Indexer's 5/30 cadence.
        executor
            .submit {
                check(replica.next() == null) { "Indexer has not accepted pending user state" }
                reconcile(snapshot(context, method = "user-state-drain"), replica)
            }.get(40, TimeUnit.SECONDS)
        awaitSaved()
        replica.close()
        closed = true
        repository.detachUserState(savedListener)
        executor.shutdown()
        pipeWriter.shutdown()
    }

    companion object {
        private const val TAG = "RiffUserState"
        private val admissions =
            java.util.concurrent.atomic
                .AtomicLong()

        fun open(context: Context): PlayerStateSession {
            val app = context.applicationContext
            val before = snapshot(app)
            val writer =
                ContentIds().id("user-state-writer") {
                    long(before.root)
                    text(app.packageName)
                    int(Process.myPid())
                    long(Process.getStartElapsedRealtime())
                    long(admissions.updateAndGet { Math.addExact(it, 1L) })
                    before.headers().forEach {
                        long(it.receipt.writerId)
                        long(it.receipt.sequence)
                    }
                }
            val initial = snapshot(app, writer, expectedRoot = before.root)
            require(initial.root == before.root) { "User-state root changed during admission" }
            return PlayerStateSession(
                app,
                UserStateReplica(
                    initial.root,
                    writer,
                    initial.installation,
                    initial.profiles,
                    initial.histories,
                    link.roadio.riff.shared.config.AndroidProjectConfiguration
                        .load(app)
                        .snapshot
                        .number("ai.selectionModel.historyDecay"),
                ),
            )
        }

        private data class Snapshot(
            val root: Long,
            val installation: InstallationStateModel,
            val profiles: Map<Int, ProfileStateModel>,
            val histories: Map<Int, ListeningHistoryModel>,
        ) {
            fun headers() =
                listOf(installation.header) + (1..3).map { profiles.getValue(it).header } + (1..3).map { histories.getValue(it).header }
        }

        private fun snapshot(
            context: Context,
            writer: Long? = null,
            method: String = "user-state",
            confirmed: Map<Pair<Int, Int>, Long> = emptyMap(),
            expectedRoot: Long? = null,
        ): Snapshot {
            val extras =
                Bundle().apply {
                    expectedRoot?.let { putLong("root", it) }
                    writer?.let { putLong("writer", it) }
                    confirmed.forEach { (key, value) ->
                        putLong("confirmed.${key.first}:${key.second}", value)
                    }
                }
            val bundle =
                checkNotNull(
                    link.roadio.riff.shared.config.ConfigurationProviderClient.call(
                        context.contentResolver,
                        Uri.parse("content://${context.packageName}.configuration"),
                        method,
                        extras,
                    ),
                )
            val descriptors =
                bundle.keySet().filter { ':' in it }.associateWith { key ->
                    @Suppress("DEPRECATION")
                    checkNotNull(bundle.getParcelable<ParcelFileDescriptor>(key))
                }
            try {
                require(descriptors.size == 7)

                fun read(key: String): java.nio.ByteBuffer {
                    val descriptor = descriptors.getValue(key)
                    val size = descriptor.statSize
                    require(size in 8..(64L * 1024 * 1024))
                    return ParcelFileDescriptor.AutoCloseInputStream(descriptor).use { stream ->
                        stream.channel.map(java.nio.channels.FileChannel.MapMode.READ_ONLY, 0L, size).asReadOnlyBuffer()
                    }
                }
                return Snapshot(
                    bundle.getLong("root"),
                    UserStateCodec.readInstallationState(read("0:0")),
                    (1..3).associateWith { UserStateCodec.readProfileState(read("1:$it")) },
                    (1..3).associateWith { UserStateCodec.readListeningHistory(read("2:$it")) },
                )
            } finally {
                descriptors.values.forEach { runCatching { it.close() } }
            }
        }

        private fun reconcile(
            snapshot: Snapshot,
            replica: UserStateReplica,
        ) {
            require(snapshot.root == replica.rootId) { "User-state root changed without fence" }
            val scopes =
                listOf(Triple(0, 0, snapshot.installation.header)) +
                    (1..3).flatMap {
                        listOf(
                            Triple(1, it, snapshot.profiles.getValue(it).header),
                            Triple(2, it, snapshot.histories.getValue(it).header),
                        )
                    }
            scopes.forEach { (kind, slot, header) ->
                if (header.receipt.writerId ==
                    replica.writerId
                ) {
                    replica.saved(snapshot.root, kind, slot, replica.writerId, header.receipt.sequence, header.generation)
                }
            }
        }
    }
}