package link.roadio.riff.userstate

import com.dynatrace.hash4j.hashing.Hashing
import link.roadio.riff.storage.publish.DirectorySync
import link.roadio.riff.userstate.model.CheckpointModel
import link.roadio.riff.userstate.model.MediaKeyModel
import link.roadio.riff.userstate.model.ProfileStateModel
import link.roadio.riff.userstate.model.QueueItemModel
import link.roadio.riff.userstate.model.QueueRefModel
import link.roadio.riff.userstate.model.QueueRevisionModel
import link.roadio.riff.userstate.model.QueueStateModel
import link.roadio.riff.userstate.model.UserStateCodec
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNull
import org.junit.Assert.assertSame
import org.junit.Assert.assertThrows
import org.junit.Assert.assertTrue
import org.junit.Assert.fail
import org.junit.Rule
import org.junit.Test
import org.junit.rules.TemporaryFolder
import java.nio.ByteBuffer
import java.nio.ByteOrder

class UserStateReplicaTest {
    @get:Rule val temporary = TemporaryFolder()
    private val hash: (ByteArray) -> Long = { Hashing.xxh3_64().hashBytesToLong(it) }

    private fun owner() = UserStateCoordinator(UserSnapshotStore(temporary.root, 10, hash, DirectorySync {}), ContentIds(hash), { 0 })

    private fun replica(owner: UserStateCoordinator) =
        UserStateReplica(
            10,
            11,
            owner.installation(),
            (1..3).associateWith(owner::profile),
            (1..3).associateWith(owner::history),
        )

    private fun start(p: ProfileStateModel) =
        p.copy(
            dynamicQueue =
                QueueRevisionModel(
                    ContentIds(hash).id("queue-revision") {
                        int(1)
                        nullable(9)
                        media(MediaKeyModel(3))
                    },
                    listOf(QueueItemModel(9, MediaKeyModel(3))),
                ).let { r ->
                    QueueStateModel(
                        definition = r,
                        execution = r,
                        checkpoint = CheckpointModel(r.id, 9, attemptId = 12, historyRecorded = true),
                    )
                },
            lastQueue = QueueRefModel(),
        )

    @Test fun acceptedStartRemainsUntilBothDurablePrefixesAndSurvivesOwnerDeath() {
        var owner = owner().also { it.openWriter(11) }
        val replica = replica(owner)
        val command = checkNotNull(replica.changeProfile(1, MediaKeyModel(3), ::start))
        owner.start(11, command.sequence, 1, checkNotNull(command.start)) { checkNotNull(command.profile) }
        replica.accepted(command)
        assertNull(replica.next())
        assertEquals(1, replica.pendingCount())
        assertFalse(command.durable.isDone)
        // Accepted RAM dies before CURRENT. The surviving replica replays the same frozen effect.
        owner = owner().also { it.openWriter(11) }
        replica.replay()
        assertSame(command, replica.next())
        owner.start(11, command.sequence, 1, checkNotNull(command.start)) { checkNotNull(command.profile) }
        owner.flush()
        replica.saved(10, 1, 1, 11, command.sequence, owner.profile(1).header.generation)
        assertFalse(command.durable.isDone)
        replica.saved(10, 2, 1, 11, command.sequence, owner.history(1).header.generation)
        assertTrue(command.durable.isDone)
        assertEquals(0, replica.pendingCount())
        assertEquals(1, replica.history(1).entries.size)
        assertEquals(1L, replica.history(1).header.generation)
    }

    @Test fun confirmedResetDiscardsOutboxButOrdinaryCloseCannot() {
        val owner = owner()
        val replica = replica(owner)
        val command = checkNotNull(replica.changeProfile(1) { it.copy(settings = it.settings.copy(theme = 1)) })
        replica.accepted(command)
        assertThrows(IllegalStateException::class.java) { replica.close() }
        assertFalse(command.durable.isDone)
        replica.discardForConfirmedReset()
        assertTrue(command.durable.isCompletedExceptionally)
        assertEquals(0, replica.pendingCount())
        assertNull(replica.next())
        assertEquals(0L, owner.profile(1).header.generation)
        replica.saved(10, 1, 1, 11, command.sequence, 1)
        assertTrue(command.durable.isCompletedExceptionally)
    }

    @Test fun staleRootAndWriterCannotAcknowledgeCurrentOperations() {
        val replica = replica(owner())
        val command = checkNotNull(replica.changeProfile(1) { it.copy(settings = it.settings.copy(theme = 1)) })
        replica.saved(20, 1, 1, 11, command.sequence, 1)
        replica.saved(10, 1, 1, 15, command.sequence, 1)
        assertFalse(command.durable.isDone)
        replica.saved(10, 1, 1, 11, command.sequence, 1)
        assertTrue(command.durable.isDone)
        assertEquals(0L, replica.history(1).header.generation)
    }

    @Test fun allTruncationsMalformedUtf8AndMissingRequiredVectorAreRejected() {
        val bytes = UserStateCodec.encode(owner().profile(1).let(::start))
        for (size in 0 until bytes.size - 4) {
            // Trailing alignment may be unused; only cuts through referenced extents must fail.
            val used = UserStateWireVerifier.verify(bytes, "ProfileState")
            if (size >= used) continue
            try {
                UserStateCodec.decodeProfileState(bytes.copyOf(size))
                fail("Truncation $size")
            } catch (_: IllegalArgumentException) {
            }
        }
        val withName =
            UserStateCodec.encode(
                owner().profile(1).let(::start).let {
                    it.copy(dynamicQueue = it.dynamicQueue.copy(name = "utf8-test"))
                },
            )
        val index = withName.toList().windowed(9).indexOfFirst { it == "utf8-test".encodeToByteArray().toList() }
        assertTrue(index > 0)
        withName[index] = 0xff.toByte()
        try {
            UserStateCodec.decodeProfileState(withName)
            fail("Invalid UTF-8")
        } catch (_: IllegalArgumentException) {
        }
        val buffer = ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN)
        val root = buffer.getInt(0)
        val vtable = root - buffer.getInt(root)
        buffer.putShort(vtable + 4 + 2 * 5, 0) // required saved_queues vector
        try {
            UserStateCodec.decodeProfileState(bytes)
            fail("Missing required vector")
        } catch (_: IllegalArgumentException) {
        }
    }
}