package com.rosetta.messenger.network

import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.CopyOnWriteArrayList
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.launch

/**
 * Centralized packet subscription registry.
 *
 * Guarantees exactly one low-level Protocol.waitPacket subscription per packet id
 * and fans out packets to:
 * 1) legacy callback listeners (waitPacket/unwaitPacket API),
 * 2) SharedFlow collectors in network/UI layers.
 */
class PacketSubscriptionRegistry(
    private val protocolProvider: () -> Protocol,
    private val scope: CoroutineScope,
    private val addLog: (String) -> Unit
) {

    private data class PacketBus(
        val packetId: Int,
        val callbacks: CopyOnWriteArrayList<(Packet) -> Unit>,
        val sharedFlow: MutableSharedFlow<Packet>,
        val protocolBridge: (Packet) -> Unit
    )

    private val buses = ConcurrentHashMap<Int, PacketBus>()

    private fun ensureBus(packetId: Int): PacketBus {
        buses[packetId]?.let { return it }

        val callbacks = CopyOnWriteArrayList<(Packet) -> Unit>()
        val sharedFlow =
            MutableSharedFlow<Packet>(
                replay = 0,
                extraBufferCapacity = 128,
                onBufferOverflow = BufferOverflow.DROP_OLDEST
            )
        val bridge: (Packet) -> Unit = { packet ->
            if (!sharedFlow.tryEmit(packet)) {
                scope.launch { sharedFlow.emit(packet) }
            }
            callbacks.forEach { callback ->
                runCatching { callback(packet) }
                    .onFailure { error ->
                        addLog("❌ PacketSubscriptionRegistry callback error: ${error.message}")
                    }
            }
        }

        val created =
            PacketBus(
                packetId = packetId,
                callbacks = callbacks,
                sharedFlow = sharedFlow,
                protocolBridge = bridge
            )

        val existing = buses.putIfAbsent(packetId, created)
        if (existing == null) {
            protocolProvider().waitPacket(packetId, bridge)
            addLog(
                "🧭 PacketSubscriptionRegistry attached id=0x${packetId.toString(16).uppercase()}"
            )
            return created
        }
        return existing
    }

    fun flow(packetId: Int): SharedFlow<Packet> = ensureBus(packetId).sharedFlow.asSharedFlow()

    fun addCallback(packetId: Int, callback: (Packet) -> Unit) {
        val bus = ensureBus(packetId)
        if (bus.callbacks.contains(callback)) {
            addLog(
                "📝 registry waitPacket(0x${packetId.toString(16)}) skipped duplicate callback; callbacks=${bus.callbacks.size}"
            )
            return
        }
        bus.callbacks.add(callback)
        addLog(
            "📝 registry waitPacket(0x${packetId.toString(16)}) callback registered; callbacks=${bus.callbacks.size}"
        )
    }

    fun removeCallback(packetId: Int, callback: (Packet) -> Unit) {
        val bus = buses[packetId] ?: return
        bus.callbacks.remove(callback)
        addLog(
            "📝 registry unwaitPacket(0x${packetId.toString(16)}) callback removed; callbacks=${bus.callbacks.size}"
        )
    }

    fun destroy() {
        buses.forEach { (packetId, bus) ->
            runCatching {
                protocolProvider().unwaitPacket(packetId, bus.protocolBridge)
            }
        }
        buses.clear()
    }
}
