package link.roadio.riff.player.storage

import link.roadio.riff.userstate.ContentIds
import link.roadio.riff.userstate.UserSnapshotStore
import link.roadio.riff.userstate.UserStateCoordinator
import link.roadio.riff.userstate.UserStateReplica
import link.roadio.riff.userstate.model.UserStateCodec
import link.roadio.riff.userstate.model.UserStateCommandModel
import link.roadio.riff.userstate.model.UserStateWriteModel
import java.io.File
import java.util.concurrent.atomic.AtomicLong

/** Test transport connects the production replica/reducer/publication core without mocking storage. */
internal class UserStateTestRig(
    directory: File,
    spacing: Long = 15L,
) {
    private val writer = writers.incrementAndGet()
    private lateinit var replica: UserStateReplica
    private val owner =
        UserStateCoordinator(
            UserSnapshotStore(directory, 71),
            ContentIds(),
            { 0L },
            bookmarkMinimumUs =
                spacing * 1_000_000L,
        ) { receipts ->
            receipts.forEach { replica.saved(71, it.kind, it.slot, it.writer, it.sequence, it.generation) }
        }
    val store: PlayerStateStore

    init {
        owner.openWriter(writer)
        replica =
            UserStateReplica(71, writer, owner.installation(), (1..3).associateWith(owner::profile), (1..3).associateWith(owner::history))
        store =
            PlayerStateStore(replica, spacing, ::deliver, ::flush, {
                flush()
                replica.close()
            })
    }

    fun flush() {
        deliver()
        owner.flush()
    }

    private fun deliver() {
        while (true) {
            val command = replica.next() ?: return
            val members = command.group
            val decoded =
                UserStateCodec.decodeUserStateCommand(
                    UserStateCodec.encode(
                        UserStateCommandModel(
                            1,
                            71,
                            writer,
                            members.map { UserStateWriteModel(it.kind, it.slot, it.sequence, it.profile, it.installation, it.start) },
                        ),
                    ),
                )
            val results =
                owner.atomic {
                    decoded.writes.map { write ->
                        when {
                            write.installation != null -> {
                                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) }
                            }
                        }
                    }
                }
            members.forEachIndexed { index, member ->
                if (results[index] == UserStateCoordinator.Result.ACCEPTED) {
                    replica.accepted(member)
                } else {
                    val header = if (member.kind == 0) owner.installation().header else owner.profile(member.slot).header
                    replica.saved(71, member.kind, member.slot, writer, member.sequence, header.generation)
                    if (member.start !=
                        null
                    ) {
                        replica.saved(71, 2, member.slot, writer, member.sequence, owner.history(member.slot).header.generation)
                    }
                }
            }
        }
    }

    companion object {
        private val writers = AtomicLong(100L)
    }
}