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.FileHeaderModel
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.assertArrayEquals
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNotEquals
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.io.File

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

    private fun store(
        root: File,
        fault: (String) -> Unit = {},
    ): UserSnapshotStore = UserSnapshotStore(root, 123, hash, DirectorySync {}, fault)

    private fun coordinator(
        root: File,
        fault: (String) -> Unit = {
        },
    ): UserStateCoordinator = UserStateCoordinator(store(root, fault), ContentIds(hash), { 0L })

    private fun qualified(
        profile: ProfileStateModel,
        media: MediaKeyModel,
        attempt: Long,
    ): ProfileStateModel {
        val item = QueueItemModel(45, media)
        val revision =
            QueueRevisionModel(
                ContentIds(hash).id("queue-revision") {
                    int(1)
                    nullable(item.occurrenceId)
                    media(item.mediaId)
                },
                listOf(item),
            )
        val queue =
            QueueStateModel(
                definition = revision,
                execution = revision,
                checkpoint = CheckpointModel(revision.id, 45, attemptId = attempt, historyRecorded = true),
            )
        return profile.copy(dynamicQueue = queue, lastQueue = QueueRefModel())
    }

    @Test fun profileWritesDoNotAgeHistoryOrOtherProfiles() {
        val root = temporary.newFolder()
        val state = coordinator(root)
        val before = checkNotNull(store(root).read())
        state.openWriter(11)
        assertEquals(
            UserStateCoordinator.Result.ACCEPTED,
            state.changeProfile(11, 1, 1) { it.copy(settings = it.settings.copy(theme = 1)) },
        )
        assertEquals(0L, state.profile(1).header.generation)
        state.flush()
        val after = checkNotNull(store(root).read())
        assertEquals(1L, state.profile(1).header.generation)
        assertEquals(0L, state.history(1).header.generation)
        before.documents.filterKeys { it != UserSnapshotStore.Key(1, 1) }.forEach { (key, bytes) ->
            assertArrayEquals(bytes, after.documents.getValue(key))
        }
    }

    @Test fun startsBatchAndDuplicateMediaRemainDistinctWithoutGrowingPastTwelveKiB() {
        val root = temporary.newFolder()
        val state = coordinator(root)
        state.openWriter(11)
        val media = MediaKeyModel(3, 0, true)
        repeat(500) { index -> state.start(11, index + 1L, 1, media) { qualified(it, media, index.toLong()) } }
        assertEquals(0L, state.history(1).header.generation)
        state.flush()
        val history = state.history(1)
        assertEquals(1L, history.header.generation)
        assertEquals(382, history.entries.size)
        assertTrue(history.entries.all { it.generation == 1L && it.mediaHi == 3L && it.mediaLo == 0L })
        val bytes = checkNotNull(store(root).read()).documents.getValue(UserSnapshotStore.Key(2, 1))
        assertEquals(12 * 1024, bytes.size)
    }

    @Test fun savedReceiptSurvivesIndexerDeathAndLostAcknowledgement() {
        val root = temporary.newFolder()
        var state = coordinator(root)
        state.openWriter(11)
        state.start(11, 1, 1, MediaKeyModel(9)) { qualified(it, MediaKeyModel(9), 10) }
        state.flush()
        state = coordinator(root)
        state.openWriter(11)
        assertEquals(UserStateCoordinator.Result.SAVED, state.start(11, 1, 1, MediaKeyModel(9)) { error("Duplicate effect") })
        state.flush()
        assertEquals(1, state.history(1).entries.size)
        assertEquals(1L, state.history(1).header.generation)
    }

    @Test fun failedCandidatePreservesAcceptedSnapshotAtEveryPrecommitBoundary() {
        for (stage in listOf("forced", "object", "manifest", "previous")) {
            val root = temporary.newFolder()
            val baseline = coordinator(root)
            val accepted = checkNotNull(store(root).read()).manifest
            var enabled = true
            val state = coordinator(root) { if (enabled && it == stage) throw java.io.IOException(stage) }
            state.openWriter(11)
            state.changeProfile(11, 1, 1) { it.copy(settings = it.settings.copy(theme = 1)) }
            try {
                state.flush()
                fail(stage)
            } catch (_: java.io.IOException) {
            }
            enabled = false
            assertEquals(accepted, checkNotNull(store(root).read()).manifest)
            assertEquals(0L, state.profile(1).header.generation)
            state.flush()
            assertEquals(1L, state.profile(1).header.generation)
            assertEquals(0L, baseline.history(1).header.generation)
        }
    }

    @Test fun failureAfterCurrentDoesNotAllocateAnotherGeneration() {
        val root = temporary.newFolder()
        coordinator(root)
        val state = coordinator(root) { if (it == "current") throw java.io.IOException("Lost completion") }
        state.openWriter(11)
        state.start(11, 1, 1, MediaKeyModel(2)) { qualified(it, MediaKeyModel(2), 10) }
        state.flush()
        state.flush()
        assertEquals(1L, state.history(1).header.generation)
        assertEquals(1, state.history(1).entries.size)
    }

    @Test fun corruptionFallsBackOnlyToPrevious() {
        val root = temporary.newFolder()
        val state = coordinator(root)
        state.openWriter(11)
        state.changeProfile(11, 1, 1) { it.copy(settings = it.settings.copy(theme = 1)) }
        state.flush()
        File(root, "CURRENT").writeBytes(byteArrayOf(0, 0))
        val restored = coordinator(root)
        assertEquals(0, restored.profile(1).settings.theme)
        File(root, "PREVIOUS").writeBytes(byteArrayOf(0, 0))
        try {
            coordinator(root)
            fail()
        } catch (_: IllegalStateException) {
        }
    }

    @Test fun recoveryPreservesDamagedCurrentBeforePublishingAndGarbageCollection() {
        val root = temporary.newFolder()
        val initial = coordinator(root)
        initial.openWriter(11)
        initial.changeProfile(11, 1, 1) { it.copy(settings = it.settings.copy(theme = 1)) }
        initial.flush()
        val damaged = byteArrayOf(1, 2)
        File(root, "CURRENT").writeBytes(damaged)
        val restored = coordinator(root)
        restored.openWriter(12)
        restored.changeProfile(12, 1, 1) { it.copy(settings = it.settings.copy(longPressStartsPlayback = true)) }
        restored.flush()
        val evidence = File(root, "quarantine").walkTopDown().filter { it.isFile && it.name == "CURRENT" }.single()
        assertArrayEquals(damaged, evidence.readBytes())
        assertTrue(
            evidence.parentFile
                .resolve("objects")
                .listFiles()
                .orEmpty()
                .isNotEmpty(),
        )
        assertTrue(coordinator(root).profile(1).settings.longPressStartsPlayback)
        assertEquals(0, coordinator(root).profile(1).settings.theme)
    }

    @Test fun unreadableCurrentIsUnavailableAndCannotAuthorizeRecoveryOrReset() {
        val root = temporary.newFolder()
        coordinator(root)
        val current = File(root, "CURRENT")
        val bytes = current.readBytes()
        assertTrue(current.setReadable(false, false))
        try {
            org.junit.Assume.assumeFalse("Filesystem permits reading mode-000 files", current.canRead())
            try {
                store(root).read()
                fail("Operational I/O must propagate")
            } catch (_: java.io.IOException) {
            }
            try {
                UserSnapshotStore.existingRootId(root, hash)
                fail("Root descriptor I/O must propagate")
            } catch (
                _: java.io.IOException,
            ) {
            }
        } finally {
            current.setReadable(true, true)
        }
        assertArrayEquals(bytes, current.readBytes())
    }

    @Test fun hashZeroAndNullableIdentifiersRoundTripSeparately() {
        val empty = QueueRevisionModel(0L)
        val profile =
            ProfileStateModel(
                header = FileHeaderModel(formatVersion = 1, profileSlot = 1),
                dynamicQueue = QueueStateModel(definition = empty, execution = empty, checkpoint = CheckpointModel(revisionId = 0L)),
                lastQueue = QueueRefModel(null),
            )
        assertEquals(profile, UserStateCodec.decodeProfileState(UserStateCodec.encode(profile)))
        assertNotEquals(profile, profile.copy(lastQueue = null))
        assertNotEquals(MediaKeyModel(1), MediaKeyModel(1, 0, true))
        UserStateValidation.profile(profile)
    }

    @Test fun attemptQualificationCannotBeAppendedTwiceOrSmuggledThroughProgress() {
        val state = coordinator(temporary.newFolder())
        state.openWriter(11)
        val media = MediaKeyModel(9)
        try {
            state.changeProfile(11, 1, 1) { qualified(it, media, 10) }
            fail()
        } catch (_: IllegalArgumentException) {
        }
        state.start(11, 1, 1, media) { qualified(it, media, 10) }
        state.start(11, 2, 1, media) { qualified(it, media, 10) }
        state.flush()
        assertEquals(1, state.history(1).entries.size)
        assertEquals(1L, state.history(1).header.generation)
    }

    @Test fun debounceHasOneAbsoluteDeadlineAndQualificationCountsPlayingTimeOnly() {
        val clock = DirtyCadence()
        clock.changed(0)
        for (now in 4000L..28000L step 4000L) clock.changed(now)
        assertEquals(30_000L, clock.deadline())
        assertFalse(clock.due(29_999))
        assertTrue(clock.due(30_000))
        clock.clear()
        clock.changed(100_000)
        assertEquals(105_000L, clock.deadline())
        val attempt = StartQualification(3000)
        assertFalse(attempt.observe(0, true))
        assertFalse(attempt.observe(1000, false))
        assertFalse(attempt.observe(60_000, true))
        assertFalse(attempt.observe(62_000, true))
        assertTrue(attempt.observe(62_001, true))
        assertFalse(attempt.observe(70_000, true))
    }
}