package link.roadio.riff.userstate

import link.roadio.riff.userstate.model.FileHeaderModel

/** Scoped delivery order, separate from entity IDs and publication generations. */
class UserStateIngress {
    private var writer: Long? = null
    private val accepted = mutableMapOf<Pair<Int, Int>, Long>()
    private val fingerprints = linkedMapOf<Triple<Int, Int, Long>, Long>()

    fun open(
        id: Long,
        headers: Map<Pair<Int, Int>, FileHeaderModel>,
        confirmed: Map<Pair<Int, Int>, Long> = emptyMap(),
    ) {
        require(confirmed.keys.all { it in headers } && confirmed.values.all { it >= 0 })
        if (writer != id) {
            accepted.clear()
            fingerprints.clear()
        }
        writer = id
        headers.forEach { (key, header) ->
            val durable = if (header.receipt.writerId == id) header.receipt.sequence else 0L
            accepted[key] = maxOf(accepted[key] ?: 0L, durable, confirmed[key] ?: 0L)
        }
    }

    fun check(
        id: Long,
        kind: Int,
        slot: Int,
        sequence: Long,
        fingerprint: Long,
    ) {
        require(writer == id && sequence > 0) { "Closed user-state writer" }
        val current = accepted.getValue(kind to slot)
        if (sequence <=
            current
        ) {
            fingerprints[
                Triple(
                    kind,
                    slot,
                    sequence,
                ),
            ]?.let { require(it == fingerprint) { "User-state sequence reused with another payload" } }
        } else {
            require(sequence == Math.addExact(current, 1L)) { "Out-of-order user-state operation" }
        }
    }

    fun applied(
        kind: Int,
        slot: Int,
        sequence: Long,
        fingerprint: Long,
    ) {
        accepted[kind to slot] = maxOf(accepted.getValue(kind to slot), sequence)
        fingerprints[Triple(kind, slot, sequence)] = fingerprint
        while (fingerprints.size > 256) fingerprints.remove(fingerprints.keys.first())
    }
}