package com.rosetta.messenger.network

import android.content.Context
import android.os.Build
import com.rosetta.messenger.data.AccountManager
import com.rosetta.messenger.data.GroupRepository
import com.rosetta.messenger.data.MessageRepository
import com.rosetta.messenger.data.isPlaceholderAccountName
import com.rosetta.messenger.utils.MessageLogger
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import java.security.SecureRandom
import java.util.*
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicInteger
import kotlin.coroutines.resume

/**
 * Singleton manager for Protocol instance
 * Ensures single connection across the app
 */
object ProtocolManager {
    private const val TAG = "ProtocolManager"
    private const val MANUAL_SYNC_BACKTRACK_MS = 120_000L
    private const val SYNC_REQUEST_TIMEOUT_MS = 12_000L
    private const val MAX_DEBUG_LOGS = 600
    private const val DEBUG_LOG_FLUSH_DELAY_MS = 60L
    private const val HEARTBEAT_OK_LOG_MIN_INTERVAL_MS = 5_000L
    private const val TYPING_INDICATOR_TIMEOUT_MS = 3_000L
    private const val PACKET_SIGNAL_PEER = 0x1A
    private const val PACKET_WEB_RTC = 0x1B
    private const val PACKET_ICE_SERVERS = 0x1C
    
    // Desktop parity: use the same primary WebSocket endpoint as desktop client.
    private const val SERVER_ADDRESS = "wss://wss.rosetta.im"
    private const val DEVICE_PREFS = "rosetta_protocol"
    private const val DEVICE_ID_KEY = "device_id"
    private const val DEVICE_ID_LENGTH = 128
    
    private var protocol: Protocol? = null
    private var messageRepository: MessageRepository? = null
    private var groupRepository: GroupRepository? = null
    private var appContext: Context? = null
    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
    @Volatile private var packetHandlersRegistered = false
    @Volatile private var stateMonitoringStarted = false
    @Volatile private var syncRequestInFlight = false
    @Volatile private var syncRequestTimeoutJob: Job? = null

    // Guard: prevent duplicate FCM token subscribe within a single session
    @Volatile
    private var lastSubscribedToken: String? = null
    
    private val _debugLogs = MutableStateFlow<List<String>>(emptyList())
    val debugLogs: StateFlow<List<String>> = _debugLogs.asStateFlow()
    private val debugLogsBuffer = ArrayDeque<String>(MAX_DEBUG_LOGS)
    private val debugLogsLock = Any()
    @Volatile private var debugFlushJob: Job? = null
    private val debugFlushPending = AtomicBoolean(false)
    @Volatile private var lastHeartbeatOkLogAtMs: Long = 0L
    @Volatile private var suppressedHeartbeatOkLogs: Int = 0
    
    // Typing status
    private val _typingUsers = MutableStateFlow<Set<String>>(emptySet())
    val typingUsers: StateFlow<Set<String>> = _typingUsers.asStateFlow()
    private val typingStateLock = Any()
    private val typingUsersByDialog = mutableMapOf<String, MutableSet<String>>()
    private val typingTimeoutJobs = ConcurrentHashMap<String, Job>()

    // Connected devices and pending verification requests
    private val _devices = MutableStateFlow<List<DeviceEntry>>(emptyList())
    val devices: StateFlow<List<DeviceEntry>> = _devices.asStateFlow()

    private val _pendingDeviceVerification = MutableStateFlow<DeviceEntry?>(null)
    val pendingDeviceVerification: StateFlow<DeviceEntry?> = _pendingDeviceVerification.asStateFlow()

    // Сигнал обновления own profile (username/name загружены с сервера)
    private val _ownProfileUpdated = MutableStateFlow(0L)
    val ownProfileUpdated: StateFlow<Long> = _ownProfileUpdated.asStateFlow()
    
    // 🔍 Global user info cache (like Desktop's InformationProvider.cachedUsers)
    // publicKey → SearchUser (resolved via PacketSearch 0x03)
    private val userInfoCache = ConcurrentHashMap<String, SearchUser>()
    // Pending resolves: publicKey → list of continuations waiting for the result
    private val pendingResolves = ConcurrentHashMap<String, MutableList<kotlinx.coroutines.CancellableContinuation<SearchUser?>>>()
    // Pending search requests: query(username/publicKey fragment) → waiting continuations
    private val pendingSearchQueries =
        ConcurrentHashMap<String, MutableList<kotlinx.coroutines.CancellableContinuation<List<SearchUser>>>>()

    private fun normalizeSearchQuery(value: String): String =
        value.trim().removePrefix("@").lowercase(Locale.ROOT)
    
    // Keep heavy protocol/message UI logs disabled by default.
    private var uiLogsEnabled = false
    private var lastProtocolState: ProtocolState? = null
    @Volatile private var syncBatchInProgress = false
    private val _syncInProgress = MutableStateFlow(false)
    val syncInProgress: StateFlow<Boolean> = _syncInProgress.asStateFlow()
    @Volatile private var resyncRequiredAfterAccountInit = false
    @Volatile private var lastForegroundSyncTime = 0L
    // Desktop parity: sequential task queue matching dialogQueue.ts (promise chain).
    // Uses Channel to guarantee strict FIFO ordering (Mutex+lastInboundJob had a race
    // condition: Dispatchers.IO doesn't guarantee FIFO, so the last-launched job could
    // finish before earlier ones, causing whenInboundTasksFinish to return prematurely
    // and BATCH_END to advance the sync timestamp while messages were still processing).
    private val inboundTaskChannel = Channel<suspend () -> Unit>(Channel.UNLIMITED)
    // Tracks the tail of the sequential processing chain (like desktop's `tail` promise)
    @Volatile private var inboundQueueDrainJob: Job? = null
    private val inboundProcessingFailures = AtomicInteger(0)
    private val syncBatchEndMutex = Mutex()

    // iOS parity: outgoing message retry mechanism.
    // iOS retries stuck WAITING messages every 4 seconds, up to 3 attempts,
    // with a hard timeout of 80 seconds. Without this, messages stay as "clocks"
    // until the next reconnect (which may never happen if the connection is alive).
    private const val OUTGOING_RETRY_INTERVAL_MS = 4_000L
    private const val OUTGOING_MAX_RETRY_ATTEMPTS = 3
    private const val OUTGOING_MAX_LIFETIME_MS = 80_000L
    private val pendingOutgoingPackets = ConcurrentHashMap<String, PacketMessage>()
    private val pendingOutgoingAttempts = ConcurrentHashMap<String, Int>()
    private val pendingOutgoingRetryJobs = ConcurrentHashMap<String, Job>()

    private fun setSyncInProgress(value: Boolean) {
        syncBatchInProgress = value
        if (_syncInProgress.value != value) {
            _syncInProgress.value = value
        }
    }
    
    fun addLog(message: String) {
        if (!uiLogsEnabled) return
        var normalizedMessage = message
        val isHeartbeatOk = normalizedMessage.startsWith("💓 Heartbeat OK")
        if (isHeartbeatOk) {
            val now = System.currentTimeMillis()
            if (now - lastHeartbeatOkLogAtMs < HEARTBEAT_OK_LOG_MIN_INTERVAL_MS) {
                suppressedHeartbeatOkLogs++
                return
            }
            if (suppressedHeartbeatOkLogs > 0) {
                normalizedMessage = "$normalizedMessage (+${suppressedHeartbeatOkLogs} skipped)"
                suppressedHeartbeatOkLogs = 0
            }
            lastHeartbeatOkLogAtMs = now
        }
        val timestamp =
                java.text.SimpleDateFormat("HH:mm:ss.SSS", Locale.getDefault()).format(Date())
        val line = "[$timestamp] $normalizedMessage"
        synchronized(debugLogsLock) {
            if (debugLogsBuffer.size >= MAX_DEBUG_LOGS) {
                debugLogsBuffer.removeFirst()
            }
            debugLogsBuffer.addLast(line)
        }
        flushDebugLogsThrottled()
    }

    fun enableUILogs(enabled: Boolean) {
        uiLogsEnabled = enabled
        MessageLogger.setEnabled(enabled)
        if (enabled) {
            val snapshot = synchronized(debugLogsLock) { debugLogsBuffer.toList() }
            _debugLogs.value = snapshot
        } else {
            _debugLogs.value = emptyList()
        }
    }

    fun clearLogs() {
        synchronized(debugLogsLock) {
            debugLogsBuffer.clear()
        }
        suppressedHeartbeatOkLogs = 0
        lastHeartbeatOkLogAtMs = 0L
        _debugLogs.value = emptyList()
    }

    private fun flushDebugLogsThrottled() {
        debugFlushPending.set(true)
        if (debugFlushJob?.isActive == true) return
        debugFlushJob =
                scope.launch {
                    while (debugFlushPending.getAndSet(false)) {
                        delay(DEBUG_LOG_FLUSH_DELAY_MS)
                        val snapshot = synchronized(debugLogsLock) { debugLogsBuffer.toList() }
                        _debugLogs.value = snapshot
                    }
                }
    }

    private fun markInboundProcessingFailure(reason: String, error: Throwable? = null) {
        inboundProcessingFailures.incrementAndGet()
        if (error != null) {
            android.util.Log.e(TAG, reason, error)
            addLog("❌ $reason: ${error.message ?: error.javaClass.simpleName}")
        } else {
            android.util.Log.w(TAG, reason)
            addLog("⚠️ $reason")
        }
    }
    
    /**
     * Инициализация с контекстом для доступа к MessageRepository
     */
    fun initialize(context: Context) {
        appContext = context.applicationContext
        messageRepository = MessageRepository.getInstance(context)
        groupRepository = GroupRepository.getInstance(context)
        if (!packetHandlersRegistered) {
            setupPacketHandlers()
            packetHandlersRegistered = true
        }
        if (!stateMonitoringStarted) {
            setupStateMonitoring()
            stateMonitoringStarted = true
        }
    }
    
    /**
     * 🔍 Мониторинг состояния соединения
     */
    private fun setupStateMonitoring() {
        scope.launch {
            getProtocol().state.collect { newState ->
                val previous = lastProtocolState
                if (newState == ProtocolState.AUTHENTICATED && previous != ProtocolState.AUTHENTICATED) {
                    // New authenticated websocket session: always allow fresh push subscribe.
                    lastSubscribedToken = null
                    onAuthenticated()
                }
                if (newState != ProtocolState.AUTHENTICATED && newState != ProtocolState.HANDSHAKING) {
                    syncRequestInFlight = false
                    clearSyncRequestTimeout()
                    setSyncInProgress(false)
                    // Connection/session dropped: force re-subscribe on next AUTHENTICATED.
                    lastSubscribedToken = null
                    // iOS parity: cancel all pending outgoing retries on disconnect.
                    // They will be retried via retryWaitingMessages() on next handshake.
                    cancelAllOutgoingRetries()
                }
                lastProtocolState = newState
            }
        }
    }
    
    /**
     * 🔥 Инициализация аккаунта - КРИТИЧНО для получения сообщений!
     * Должен вызываться после авторизации пользователя
     */
    fun initializeAccount(publicKey: String, privateKey: String) {
        setSyncInProgress(false)
        clearTypingState()
        messageRepository?.initialize(publicKey, privateKey)
        if (resyncRequiredAfterAccountInit || protocol?.isAuthenticated() == true) {
            resyncRequiredAfterAccountInit = false
            requestSynchronize()
        }
        // Send "Rosetta Updates" message on version change (like desktop useUpdateMessage)
        scope.launch {
            messageRepository?.checkAndSendVersionUpdateMessage()
        }
    }
    
    /**
     * Настройка обработчиков пакетов
     */
    private fun setupPacketHandlers() {
        // Обработчик входящих сообщений (0x06)
        // Desktop parity: desktop client does not send PacketDelivery manually.
        waitPacket(0x06) { packet ->
            val messagePacket = packet as PacketMessage
            
            launchInboundPacketTask {
                val repository = messageRepository
                if (repository == null || !repository.isInitialized()) {
                    requireResyncAfterAccountInit("⏳ Incoming message before account init, scheduling re-sync")
                    markInboundProcessingFailure("Incoming packet skipped before account init")
                    return@launchInboundPacketTask
                }
                val processed = repository.handleIncomingMessage(messagePacket)
                if (!processed) {
                    markInboundProcessingFailure(
                        "Message processing failed for ${messagePacket.messageId.take(8)}"
                    )
                    return@launchInboundPacketTask
                }
                if (!syncBatchInProgress) {
                    repository.updateLastSyncTimestamp(messagePacket.timestamp)
                }
            }
        }
        
        // Обработчик доставки (0x08)
        // Desktop parity: useDialogFiber.ts updates sync time on delivery (await updateSyncTime(Date.now()))
        waitPacket(0x08) { packet ->
            val deliveryPacket = packet as PacketDelivery
            
            launchInboundPacketTask {
                val repository = messageRepository
                if (repository == null || !repository.isInitialized()) {
                    requireResyncAfterAccountInit("⏳ Delivery status before account init, scheduling re-sync")
                    markInboundProcessingFailure("Delivery packet skipped before account init")
                    return@launchInboundPacketTask
                }
                try {
                    repository.handleDelivery(deliveryPacket)
                    // iOS parity: cancel retry timer on delivery ACK
                    resolveOutgoingRetry(deliveryPacket.messageId)
                } catch (e: Exception) {
                    markInboundProcessingFailure("Delivery processing failed", e)
                    return@launchInboundPacketTask
                }
                if (!syncBatchInProgress) {
                    repository.updateLastSyncTimestamp(System.currentTimeMillis())
                }
            }
        }
        
        // Обработчик прочтения (0x07)
        // В Desktop PacketRead не содержит messageId - сообщает что собеседник прочитал сообщения
        waitPacket(0x07) { packet ->
            val readPacket = packet as PacketRead
            
            launchInboundPacketTask {
                val repository = messageRepository
                if (repository == null || !repository.isInitialized()) {
                    requireResyncAfterAccountInit("⏳ Read status before account init, scheduling re-sync")
                    markInboundProcessingFailure("Read packet skipped before account init")
                    return@launchInboundPacketTask
                }
                val ownKey = getProtocol().getPublicKey().orEmpty()
                if (ownKey.isBlank()) {
                    requireResyncAfterAccountInit("⏳ Read status before protocol account init, scheduling re-sync")
                    markInboundProcessingFailure("Read packet skipped before protocol account init")
                    return@launchInboundPacketTask
                }
                try {
                    repository.handleRead(readPacket)
                } catch (e: Exception) {
                    markInboundProcessingFailure("Read processing failed", e)
                    return@launchInboundPacketTask
                }
                if (!syncBatchInProgress) {
                    // Desktop parity:
                    // own direct read sync (from=me,to=peer) does not advance sync cursor.
                    val isOwnDirectReadSync =
                            readPacket.fromPublicKey.trim() == ownKey &&
                                    !isGroupDialogKey(readPacket.toPublicKey)
                    if (!isOwnDirectReadSync) {
                        repository.updateLastSyncTimestamp(System.currentTimeMillis())
                    }
                }
            }
        }

        // 🔐 New device login attempt (0x09) — desktop parity (system Safe message)
        waitPacket(0x09) { packet ->
            val devicePacket = packet as PacketDeviceNew

            addLog(
                "🔐 DEVICE LOGIN ATTEMPT: ip=${devicePacket.ipAddress}, device=${devicePacket.device.deviceName}, os=${devicePacket.device.deviceOs}"
            )

            launchInboundPacketTask {
                messageRepository?.addDeviceLoginSystemMessage(
                    ipAddress = devicePacket.ipAddress,
                    deviceId = devicePacket.device.deviceId,
                    deviceName = devicePacket.device.deviceName,
                    deviceOs = devicePacket.device.deviceOs
                )
            }
        }

        // 🔄 Обработчик батчевой синхронизации (0x19)
        waitPacket(0x19) { packet ->
            handleSyncPacket(packet as PacketSync)
        }

        // 👥 Обработчик синхронизации групп (0x14)
        // Desktop parity: во время sync сервер отправляет PacketGroupJoin с groupString.
        waitPacket(0x14) { packet ->
            val joinPacket = packet as PacketGroupJoin

            launchInboundPacketTask {
                val repository = messageRepository
                val groups = groupRepository
                val account = repository?.getCurrentAccountKey()
                val privateKey = repository?.getCurrentPrivateKey()
                if (groups == null || account.isNullOrBlank() || privateKey.isNullOrBlank()) {
                    return@launchInboundPacketTask
                }
                try {
                    val result = groups.synchronizeJoinedGroup(
                        accountPublicKey = account,
                        accountPrivateKey = privateKey,
                        packet = joinPacket
                    )
                    if (result?.success == true) {
                        addLog("👥 GROUP synced: ${result.dialogPublicKey}")
                    }
                } catch (e: Exception) {
                    android.util.Log.w(TAG, "Failed to sync group packet", e)
                }
            }
        }
        
        // 🟢 Обработчик онлайн-статуса (0x05)
        waitPacket(0x05) { packet ->
            val onlinePacket = packet as PacketOnlineState
            
            scope.launch {
                if (messageRepository != null) {
                    onlinePacket.publicKeysState.forEach { item ->
                        val isOnline = item.state == OnlineState.ONLINE
                        messageRepository?.updateOnlineStatus(item.publicKey, isOnline)
                    }
                }
            }
        }
        
        // Обработчик typing (0x0B)
        waitPacket(0x0B) { packet ->
            val typingPacket = packet as PacketTyping
            val fromPublicKey = typingPacket.fromPublicKey.trim()
            val toPublicKey = typingPacket.toPublicKey.trim()
            if (fromPublicKey.isBlank() || toPublicKey.isBlank()) return@waitPacket

            val ownPublicKey =
                getProtocol().getPublicKey()?.trim().orEmpty().ifBlank {
                    messageRepository?.getCurrentAccountKey()?.trim().orEmpty()
                }
            if (ownPublicKey.isNotBlank() && fromPublicKey.equals(ownPublicKey, ignoreCase = true)) {
                return@waitPacket
            }

            val dialogKey =
                resolveTypingDialogKey(
                    fromPublicKey = fromPublicKey,
                    toPublicKey = toPublicKey,
                    ownPublicKey = ownPublicKey
                ) ?: return@waitPacket

            rememberTypingEvent(dialogKey, fromPublicKey)
        }

        // 📱 Обработчик списка устройств (0x17)
        waitPacket(0x17) { packet ->
            val devicesPacket = packet as PacketDeviceList
            val parsedDevices = devicesPacket.devices
            _devices.value = parsedDevices
            _pendingDeviceVerification.value = parsedDevices.firstOrNull { device ->
                device.deviceVerify == DeviceVerifyState.NOT_VERIFIED
            }
        }
        
        // 🔥 Обработчик поиска/user info (0x03)
        // Обновляет информацию о пользователе в диалогах когда приходит ответ от сервера
        // + обновляет own profile (username/name) аналогично Desktop useUserInformation()
        waitPacket(0x03) { packet ->
            val searchPacket = packet as PacketSearch
            
            if (searchPacket.users.isNotEmpty()) {
                scope.launch(Dispatchers.IO) {
                    val ownPublicKey = getProtocol().getPublicKey()
                    searchPacket.users.forEach { user ->
                        // 🔍 Кэшируем всех пользователей (desktop parity: cachedUsers)
                        userInfoCache[user.publicKey] = user
                        
                        // Resume pending resolves for this publicKey
                        pendingResolves.remove(user.publicKey)?.forEach { cont ->
                            try { cont.resume(user) } catch (_: Exception) {}
                        }
                        
                        // Обновляем инфо в диалогах (для всех пользователей)
                        messageRepository?.updateDialogUserInfo(
                            user.publicKey,
                            user.title,
                            user.username,
                            user.verified
                        )
                        
                        // Если это наш own profile — сохраняем username/name в AccountManager
                        if (user.publicKey == ownPublicKey && appContext != null) {
                            val accountManager = AccountManager(appContext!!)
                            if (user.title.isNotBlank() && !isPlaceholderAccountName(user.title)) {
                                accountManager.updateAccountName(user.publicKey, user.title)
                            }
                            if (user.username.isNotBlank()) {
                                accountManager.updateAccountUsername(user.publicKey, user.username)
                            }
                            _ownProfileUpdated.value = System.currentTimeMillis()
                        }
                    }
                }
            }
            
            // Resume pending resolves that got empty response (no match)
            if (searchPacket.search.isNotEmpty() && searchPacket.users.none { it.publicKey == searchPacket.search }) {
                pendingResolves.remove(searchPacket.search)?.forEach { cont ->
                    try { cont.resume(null) } catch (_: Exception) {}
                }
            }

            // Resume pending username/query searches.
            // Server may return query in different case/format, so match robustly.
            if (searchPacket.search.isNotEmpty()) {
                val rawQuery = searchPacket.search.trim()
                val normalizedQuery = normalizeSearchQuery(rawQuery)
                val continuations = LinkedHashSet<kotlinx.coroutines.CancellableContinuation<List<SearchUser>>>()

                fun collectByKey(key: String) {
                    if (key.isEmpty()) return
                    pendingSearchQueries.remove(key)?.let { continuations.addAll(it) }
                }

                collectByKey(rawQuery)
                if (normalizedQuery.isNotEmpty() && normalizedQuery != rawQuery) {
                    collectByKey(normalizedQuery)
                }

                if (continuations.isEmpty()) {
                    val matchedByQuery =
                        pendingSearchQueries.keys.firstOrNull { pendingKey ->
                            pendingKey.equals(rawQuery, ignoreCase = true) ||
                                    normalizeSearchQuery(pendingKey) == normalizedQuery
                        }
                    if (matchedByQuery != null) collectByKey(matchedByQuery)
                }

                if (continuations.isEmpty() && searchPacket.users.isNotEmpty()) {
                    val responseUsernames =
                        searchPacket.users
                            .map { normalizeSearchQuery(it.username) }
                            .filter { it.isNotEmpty() }
                            .toSet()
                    if (responseUsernames.isNotEmpty()) {
                        val matchedByUsers =
                            pendingSearchQueries.keys.firstOrNull { pendingKey ->
                                val normalizedPending = normalizeSearchQuery(pendingKey)
                                normalizedPending.isNotEmpty() &&
                                        responseUsernames.contains(normalizedPending)
                            }
                        if (matchedByUsers != null) collectByKey(matchedByUsers)
                    }
                }

                continuations.forEach { cont ->
                    try {
                        cont.resume(searchPacket.users)
                    } catch (_: Exception) {}
                }
            }
        }
        
        // 🚀 Обработчик транспортного сервера (0x0F)
        waitPacket(0x0F) { packet ->
            val transportPacket = packet as PacketRequestTransport
            TransportManager.setTransportServer(transportPacket.transportServer)
        }
    }

    /**
     * Desktop parity: sequential task queue (like dialogQueue.ts runTaskInQueue / whenFinish).
     *
     * Desktop uses a promise chain: `tail = tail.then(fn).catch(...)` which guarantees
     * strict FIFO ordering and `whenFinish = () => tail` returns a promise that resolves
     * only after ALL queued tasks complete.
     *
     * We reproduce this with a Channel<suspend () -> Unit> (UNLIMITED buffer) consumed
     * by a single coroutine. Tasks are executed strictly in the order they were submitted,
     * and `whenInboundTasksFinish()` waits for the queue to drain completely.
     */
    private fun ensureInboundQueueDrainRunning() {
        if (inboundQueueDrainJob?.isActive == true) return
        inboundQueueDrainJob = scope.launch {
            for (task in inboundTaskChannel) {
                try {
                    task()
                } catch (t: Throwable) {
                    markInboundProcessingFailure("Dialog queue error", t)
                }
            }
        }
    }

    private fun launchInboundPacketTask(block: suspend () -> Unit): Boolean {
        ensureInboundQueueDrainRunning()
        val result = inboundTaskChannel.trySend(block)
        if (result.isFailure) {
            markInboundProcessingFailure(
                    "Failed to enqueue inbound task",
                    result.exceptionOrNull()
            )
            return false
        }
        return true
    }

    private fun requireResyncAfterAccountInit(reason: String) {
        if (!resyncRequiredAfterAccountInit) {
            addLog(reason)
        }
        resyncRequiredAfterAccountInit = true
    }

    /**
     * Desktop parity: equivalent of `await whenFinish()` in useSynchronize.ts.
     * Sends a sentinel task into the sequential queue and suspends until it executes.
     * Since the queue is strictly FIFO, when the sentinel runs, all previously
     * submitted tasks are guaranteed to have completed.
     */
    private suspend fun whenInboundTasksFinish(): Boolean {
        val done = CompletableDeferred<Unit>()
        if (!launchInboundPacketTask { done.complete(Unit) }) {
            return false
        }
        done.await()
        return true
    }

    private fun isGroupDialogKey(value: String): Boolean {
        val normalized = value.trim().lowercase(Locale.ROOT)
        return normalized.startsWith("#group:") || normalized.startsWith("group:")
    }

    private fun normalizeGroupDialogKey(value: String): String {
        val trimmed = value.trim()
        val normalized = trimmed.lowercase(Locale.ROOT)
        return when {
            normalized.startsWith("#group:") -> "#group:${trimmed.substringAfter(':').trim()}"
            normalized.startsWith("group:") -> "#group:${trimmed.substringAfter(':').trim()}"
            else -> trimmed
        }
    }

    private fun resolveTypingDialogKey(
        fromPublicKey: String,
        toPublicKey: String,
        ownPublicKey: String
    ): String? {
        return when {
            isGroupDialogKey(toPublicKey) -> normalizeGroupDialogKey(toPublicKey)
            ownPublicKey.isNotBlank() && toPublicKey.equals(ownPublicKey, ignoreCase = true) ->
                fromPublicKey.trim()
            else -> null
        }
    }

    private fun makeTypingTimeoutKey(dialogKey: String, fromPublicKey: String): String {
        return "${dialogKey.lowercase(Locale.ROOT)}|${fromPublicKey.lowercase(Locale.ROOT)}"
    }

    private fun rememberTypingEvent(dialogKey: String, fromPublicKey: String) {
        val normalizedDialogKey =
            if (isGroupDialogKey(dialogKey)) normalizeGroupDialogKey(dialogKey) else dialogKey.trim()
        val normalizedFrom = fromPublicKey.trim()
        if (normalizedDialogKey.isBlank() || normalizedFrom.isBlank()) return

        synchronized(typingStateLock) {
            val users = typingUsersByDialog.getOrPut(normalizedDialogKey) { mutableSetOf() }
            users.add(normalizedFrom)
            _typingUsers.value = typingUsersByDialog.keys.toSet()
        }

        val timeoutKey = makeTypingTimeoutKey(normalizedDialogKey, normalizedFrom)
        typingTimeoutJobs.remove(timeoutKey)?.cancel()
        typingTimeoutJobs[timeoutKey] =
            scope.launch {
                delay(TYPING_INDICATOR_TIMEOUT_MS)
                synchronized(typingStateLock) {
                    val users = typingUsersByDialog[normalizedDialogKey]
                    users?.remove(normalizedFrom)
                    if (users.isNullOrEmpty()) {
                        typingUsersByDialog.remove(normalizedDialogKey)
                    }
                    _typingUsers.value = typingUsersByDialog.keys.toSet()
                }
                typingTimeoutJobs.remove(timeoutKey)
            }
    }

    private fun clearTypingState() {
        typingTimeoutJobs.values.forEach { it.cancel() }
        typingTimeoutJobs.clear()
        synchronized(typingStateLock) {
            typingUsersByDialog.clear()
            _typingUsers.value = emptySet()
        }
    }

    private fun onAuthenticated() {
        setSyncInProgress(false)
        TransportManager.requestTransportServer()
        com.rosetta.messenger.update.UpdateManager.requestSduServer()
        fetchOwnProfile()
        requestSynchronize()
        subscribePushTokenIfAvailable()
    }

    private fun finishSyncCycle(reason: String) {
        syncRequestInFlight = false
        clearSyncRequestTimeout()
        inboundProcessingFailures.set(0)
        addLog(reason)
        setSyncInProgress(false)
        retryWaitingMessages()
        requestMissingUserInfo()
    }

    /**
     * Send FCM push token to server (SUBSCRIBE).
     * Deduplicates: won't re-send the same token within one connection session.
     * Called internally on AUTHENTICATED and can be called from
     * [com.rosetta.messenger.push.RosettaFirebaseMessagingService.onNewToken]
     * when Firebase rotates the token mid-session.
     *
     * @param forceToken  if non-null, use this token instead of reading SharedPreferences
     *                    (used by onNewToken which already has the fresh token).
     */
    fun subscribePushTokenIfAvailable(forceToken: String? = null) {
        val context = appContext ?: return
        val token = (forceToken
                ?: context.getSharedPreferences("rosetta_prefs", Context.MODE_PRIVATE)
                        .getString("fcm_token", null))
                ?.trim()
                .orEmpty()
        if (token.isEmpty()) return

        // Dedup: don't send the same token twice in one connection session
        if (token == lastSubscribedToken) {
            addLog("🔔 Push token already subscribed this session — skipped")
            return
        }

        val subPacket = PacketPushNotification().apply {
            notificationsToken = token
            action = PushNotificationAction.SUBSCRIBE
        }
        send(subPacket)
        lastSubscribedToken = token
        addLog("🔔 Push token SUBSCRIBE sent")
    }

    private fun requestSynchronize() {
        if (syncBatchInProgress) {
            addLog("⚠️ SYNC request skipped: sync already in progress")
            return
        }
        if (syncRequestInFlight) {
            addLog("⚠️ SYNC request skipped: previous request still in flight")
            return
        }
        syncRequestInFlight = true
        addLog("🔄 SYNC requested — fetching last sync timestamp...")
        scope.launch {
            val repository = messageRepository
            if (repository == null || !repository.isInitialized()) {
                syncRequestInFlight = false
                clearSyncRequestTimeout()
                requireResyncAfterAccountInit("⏳ Sync postponed until account is initialized")
                return@launch
            }
            val protocolAccount = getProtocol().getPublicKey().orEmpty()
            val repositoryAccount = repository.getCurrentAccountKey().orEmpty()
            if (protocolAccount.isNotBlank() &&
                            repositoryAccount.isNotBlank() &&
                            repositoryAccount != protocolAccount
            ) {
                syncRequestInFlight = false
                clearSyncRequestTimeout()
                requireResyncAfterAccountInit(
                        "⏳ Sync postponed: repository bound to another account"
                )
                return@launch
            }
            val lastSync = repository.getLastSyncTimestamp()
            addLog("🔄 SYNC sending request with lastSync=$lastSync")
            sendSynchronize(lastSync)
        }
    }

    private fun sendSynchronize(timestamp: Long) {
        syncRequestInFlight = true
        scheduleSyncRequestTimeout(timestamp)
        val packet = PacketSync().apply {
            status = SyncStatus.NOT_NEEDED
            this.timestamp = timestamp
        }
        send(packet)
    }

    /**
     * Desktop parity: useSynchronize.ts usePacket(25, ...)
     * BATCH_START → mark sync in progress (synchronous — no scope.launch)
     * BATCH_END → wait for ALL message tasks to finish, save timestamp, request next batch
     * NOT_NEEDED → sync complete, mark connected (synchronous — no scope.launch)
     *
     * CRITICAL: BATCH_START and NOT_NEEDED are handled synchronously in the WebSocket
     * callback thread. This prevents a race condition where 0x06 message packets arrive
     * and check syncBatchInProgress BEFORE the scope.launch coroutine for BATCH_START
     * has been scheduled on Dispatchers.IO.
     */
    private fun handleSyncPacket(packet: PacketSync) {
        syncRequestInFlight = false
        clearSyncRequestTimeout()
        when (packet.status) {
            SyncStatus.BATCH_START -> {
                addLog("🔄 SYNC BATCH_START — incoming message batch")
                // Synchronous — guarantees syncBatchInProgress=true before any
                // subsequent 0x06 packets are dispatched by OkHttp's sequential callback.
                setSyncInProgress(true)
                inboundProcessingFailures.set(0)
            }
            SyncStatus.BATCH_END -> {
                addLog("🔄 SYNC BATCH_END — waiting for tasks to finish (ts=${packet.timestamp})")
                // BATCH_END requires suspend (whenInboundTasksFinish), so we launch a coroutine.
                // syncBatchInProgress stays true until NOT_NEEDED arrives.
                scope.launch {
                    syncBatchEndMutex.withLock {
                        val tasksFinished = whenInboundTasksFinish()
                        if (!tasksFinished) {
                            android.util.Log.w(
                                TAG,
                                "SYNC BATCH_END: queue unavailable, skipping cursor update for this step"
                            )
                            val fallbackCursor = messageRepository?.getLastSyncTimestamp() ?: 0L
                            sendSynchronize(fallbackCursor)
                            return@launch
                        }
                        val failuresInBatch = inboundProcessingFailures.getAndSet(0)
                        if (failuresInBatch > 0) {
                            addLog(
                                "⚠️ SYNC batch had $failuresInBatch processing error(s), continuing with desktop sync cursor behavior"
                            )
                        }
                        val repository = messageRepository
                        // Desktop parity: save the cursor provided by BATCH_END and request next
                        // chunk with the same cursor.
                        repository?.updateLastSyncTimestamp(packet.timestamp)
                        addLog("🔄 SYNC tasks done — cursor=${packet.timestamp}, requesting next batch")
                        sendSynchronize(packet.timestamp)
                    }
                }
            }
            SyncStatus.NOT_NEEDED -> {
                // Synchronous — immediately marks sync as complete.
                // Desktop parity: NOT_NEEDED just sets state to CONNECTED,
                // does NOT update last_sync timestamp (unnecessary since client
                // was already up to date).
                finishSyncCycle("✅ SYNC COMPLETE — no more messages to sync")
            }
        }
    }

    private fun scheduleSyncRequestTimeout(cursor: Long) {
        syncRequestTimeoutJob?.cancel()
        syncRequestTimeoutJob = scope.launch {
            delay(SYNC_REQUEST_TIMEOUT_MS)
            if (!syncRequestInFlight || !isAuthenticated()) return@launch
            syncRequestInFlight = false
            addLog(
                    "⏱️ SYNC response timeout for cursor=$cursor, retrying request"
            )
            requestSynchronize()
        }
    }

    private fun clearSyncRequestTimeout() {
        syncRequestTimeoutJob?.cancel()
        syncRequestTimeoutJob = null
    }

    /**
     * Retry messages stuck in WAITING status on reconnect.
     * Desktop has in-memory _packetQueue that flushes on handshake, but desktop apps are
     * rarely force-killed. On Android, the app can be killed mid-send, leaving messages
     * in WAITING status in the DB. This method resends them after sync completes.
     *
     * Messages older than 80s (MESSAGE_MAX_TIME_TO_DELEVERED_S) are marked ERROR.
     */
    private fun retryWaitingMessages() {
        scope.launch {
            val repository = messageRepository
            if (repository == null || !repository.isInitialized()) return@launch
            try {
                repository.retryWaitingMessages()
            } catch (e: Exception) {
                android.util.Log.e(TAG, "retryWaitingMessages failed", e)
            }
        }
    }

    /**
     * Desktop parity: after sync completes, resolve names/usernames for all dialogs
     * that still have empty titles. Clears the one-shot guard first so that previously
     * failed requests can be retried.
     */
    private fun requestMissingUserInfo() {
        scope.launch {
            val repository = messageRepository
            if (repository == null || !repository.isInitialized()) return@launch
            try {
                repository.clearUserInfoRequestCache()
                repository.requestMissingUserInfo()
            } catch (e: Exception) {
                android.util.Log.e(TAG, "requestMissingUserInfo failed", e)
            }
        }
    }
    
    /**
     * Get or create Protocol instance
     */
    fun getProtocol(): Protocol {
        if (protocol == null) {
            protocol = Protocol(SERVER_ADDRESS) { msg -> addLog(msg) }
        }
        return protocol!!
    }
    
    /**
     * Get connection state flow
     */
    val state: StateFlow<ProtocolState>
        get() = getProtocol().state
    
    /**
     * Get last error flow
     */
    val lastError: StateFlow<String?>
        get() = getProtocol().lastError
    
    /**
     * Connect to server
     */
    fun connect() {
        getProtocol().connect()
    }

    /**
     * Trigger immediate reconnect on app foreground (skip waiting backoff timer).
     */
    fun reconnectNowIfNeeded(reason: String = "foreground_resume") {
        getProtocol().reconnectNowIfNeeded(reason)
    }

    /**
     * Desktop parity: синхронизация при каждом заходе в приложение.
     * Desktop вызывает trySync() на каждый handshakeExchangeComplete.
     * На Android — вызываем из onResume(), если уже AUTHENTICATED и не идёт sync.
     * Дебаунс 5 секунд чтобы не спамить при быстром alt-tab.
     */
    fun syncOnForeground() {
        if (!isAuthenticated()) return
        if (syncBatchInProgress) return
        if (syncRequestInFlight) return
        val now = System.currentTimeMillis()
        if (now - lastForegroundSyncTime < 5_000) return
        lastForegroundSyncTime = now
        addLog("🔄 SYNC on foreground resume")
        requestSynchronize()
    }

    /**
     * Manual sync trigger from UI.
     * Rewinds lastSync a bit to safely re-fetch recent packets and re-starts sync.
     */
    fun forceSynchronize(backtrackMs: Long = MANUAL_SYNC_BACKTRACK_MS) {
        if (!isAuthenticated()) {
            reconnectNowIfNeeded("manual_sync_button")
            return
        }
        if (syncBatchInProgress) return
        if (syncRequestInFlight) return

        scope.launch {
            val repository = messageRepository
            if (repository == null || !repository.isInitialized()) {
                requireResyncAfterAccountInit("⏳ Manual sync postponed until account is initialized")
                return@launch
            }
            val currentSync = repository.getLastSyncTimestamp()
            val rewindTo = (currentSync - backtrackMs.coerceAtLeast(0L)).coerceAtLeast(0L)

            syncRequestInFlight = true
            addLog("🔄 MANUAL SYNC requested: lastSync=$currentSync -> rewind=$rewindTo")
            sendSynchronize(rewindTo)
        }
    }

    /**
     * Authenticate with server
     */
    fun authenticate(publicKey: String, privateHash: String) {
        val device = buildHandshakeDevice()
        getProtocol().startHandshake(publicKey, privateHash, device)
    }
    
    /**
     * Запрашивает собственный профиль с сервера (username, name/title).
     * Аналог Desktop: useUserInformation(ownPublicKey) → PacketSearch(0x03)
     */
    private fun fetchOwnProfile() {
        val publicKey = getProtocol().getPublicKey() ?: return
        val privateHash = getProtocol().getPrivateHash() ?: return
        
        val packet = PacketSearch().apply {
            this.privateKey = privateHash
            this.search = publicKey
        }
        send(packet)
    }
    
    /**
     * 🔍 Resolve publicKey → user title (like Desktop useUserInformation)
     * Checks cache first, then sends PacketSearch and waits for response.
     * Returns title or null on timeout/not found.
     * @param timeoutMs max wait time for server response (default 3s)
     */
    suspend fun resolveUserName(publicKey: String, timeoutMs: Long = 3000): String? {
        if (publicKey.isEmpty()) return null
        
        // 1. Check in-memory cache (instant)
        userInfoCache[publicKey]?.let { cached ->
            val name = cached.title.ifEmpty { cached.username }
            if (name.isNotEmpty()) return name
        }
        
        // 2. Send PacketSearch and wait for response via suspendCancellableCoroutine
        val privateHash = try { getProtocol().getPrivateHash() } catch (_: Exception) { null }
            ?: return null
        
        return try {
            withTimeout(timeoutMs) {
                suspendCancellableCoroutine { cont ->
                    // Register continuation in pending list
                    pendingResolves.getOrPut(publicKey) { mutableListOf() }.add(cont)
                    
                    cont.invokeOnCancellation {
                        pendingResolves[publicKey]?.remove(cont)
                        if (pendingResolves[publicKey]?.isEmpty() == true) {
                            pendingResolves.remove(publicKey)
                        }
                    }
                    
                    // Send search request
                    val packet = PacketSearch().apply {
                        this.privateKey = privateHash
                        this.search = publicKey
                    }
                    send(packet)
                }
            }?.let { user -> user.title.ifEmpty { user.username }.ifEmpty { null } }
        } catch (_: Exception) {
            // Timeout or cancellation — clean up
            pendingResolves.remove(publicKey)
            null
        }
    }
    
    /**
     * 🔍 Get cached user info (no network request)
     */
    fun getCachedUserName(publicKey: String): String? {
        val cached = userInfoCache[publicKey] ?: return null
        return cached.title.ifEmpty { cached.username }.ifEmpty { null }
    }

    /**
     * 🔍 Get full cached user info (no network request)
     */
    fun getCachedUserInfo(publicKey: String): SearchUser? {
        return userInfoCache[publicKey]
    }

    /**
     * 🔍 Get cached user by username (no network request).
     * Username compare is case-insensitive and ignores '@'.
     */
    fun getCachedUserByUsername(username: String): SearchUser? {
        val normalizedUsername = normalizeSearchQuery(username)
        if (normalizedUsername.isEmpty()) return null
        return userInfoCache.values.firstOrNull { cached ->
            normalizeSearchQuery(cached.username) == normalizedUsername
        }
    }

    /**
     * 🔍 Resolve publicKey → full SearchUser (with server request if needed)
     */
    suspend fun resolveUserInfo(publicKey: String, timeoutMs: Long = 3000): SearchUser? {
        if (publicKey.isEmpty()) return null

        // 1. Check in-memory cache
        userInfoCache[publicKey]?.let { return it }

        // 2. Send PacketSearch and wait
        val privateHash = try { getProtocol().getPrivateHash() } catch (_: Exception) { null }
            ?: return null

        return try {
            withTimeout(timeoutMs) {
                suspendCancellableCoroutine { cont ->
                    pendingResolves.getOrPut(publicKey) { mutableListOf() }.add(cont)
                    cont.invokeOnCancellation {
                        pendingResolves[publicKey]?.remove(cont)
                        if (pendingResolves[publicKey]?.isEmpty() == true) {
                            pendingResolves.remove(publicKey)
                        }
                    }
                    val packet = PacketSearch().apply {
                        this.privateKey = privateHash
                        this.search = publicKey
                    }
                    send(packet)
                }
            }
        } catch (_: Exception) {
            pendingResolves.remove(publicKey)
            null
        }
    }

    /**
     * 🔍 Search users by query (usually username without '@').
     * Returns raw PacketSearch users list for the exact query.
     */
    suspend fun searchUsers(query: String, timeoutMs: Long = 3000): List<SearchUser> {
        val normalizedQuery = normalizeSearchQuery(query)
        if (normalizedQuery.isEmpty()) return emptyList()

        val privateHash = try { getProtocol().getPrivateHash() } catch (_: Exception) { null }
            ?: return emptyList()

        val cachedMatches =
            userInfoCache.values.filter { cached ->
                normalizeSearchQuery(cached.username) == normalizedQuery && cached.publicKey.isNotBlank()
            }
        if (cachedMatches.isNotEmpty()) {
            return cachedMatches.distinctBy { it.publicKey }
        }

        return try {
            withTimeout(timeoutMs) {
                suspendCancellableCoroutine { cont ->
                    pendingSearchQueries
                        .getOrPut(normalizedQuery) { mutableListOf() }
                        .add(cont)

                    cont.invokeOnCancellation {
                        pendingSearchQueries[normalizedQuery]?.remove(cont)
                        if (pendingSearchQueries[normalizedQuery]?.isEmpty() == true) {
                            pendingSearchQueries.remove(normalizedQuery)
                        }
                    }

                    val packet = PacketSearch().apply {
                        this.privateKey = privateHash
                        this.search = normalizedQuery
                    }
                    send(packet)
                }
            }
        } catch (_: Exception) {
            pendingSearchQueries.remove(normalizedQuery)
            emptyList()
        }
    }

    /**
     * Accept a pending device login request.
     */
    fun acceptDevice(deviceId: String) {
        val packet = PacketDeviceResolve().apply {
            this.deviceId = deviceId
            this.solution = DeviceResolveSolution.ACCEPT
        }
        send(packet)
    }

    /**
     * Decline a pending device login request.
     */
    fun declineDevice(deviceId: String) {
        val packet = PacketDeviceResolve().apply {
            this.deviceId = deviceId
            this.solution = DeviceResolveSolution.DECLINE
        }
        send(packet)
    }
    
    /**
     * Send packet (simplified)
     */
    fun send(packet: Packet) {
        getProtocol().sendPacket(packet)
    }

    /**
     * Send an outgoing message packet and register it for automatic retry.
     * iOS parity: registerOutgoingRetry + scheduleOutgoingRetry.
     */
    fun sendMessageWithRetry(packet: PacketMessage) {
        send(packet)
        registerOutgoingRetry(packet)
    }

    /**
     * iOS parity: register an outgoing message for automatic retry.
     */
    private fun registerOutgoingRetry(packet: PacketMessage) {
        val messageId = packet.messageId
        pendingOutgoingRetryJobs[messageId]?.cancel()
        pendingOutgoingPackets[messageId] = packet
        pendingOutgoingAttempts[messageId] = 0
        scheduleOutgoingRetry(messageId)
    }

    /**
     * iOS parity: schedule a retry attempt for a pending outgoing message.
     * Retries every 4 seconds, marks as ERROR after 80s or 3 attempts.
     */
    private fun scheduleOutgoingRetry(messageId: String) {
        pendingOutgoingRetryJobs[messageId]?.cancel()
        pendingOutgoingRetryJobs[messageId] = scope.launch {
            delay(OUTGOING_RETRY_INTERVAL_MS)

            val packet = pendingOutgoingPackets[messageId] ?: return@launch
            val attempts = pendingOutgoingAttempts[messageId] ?: 0

            // Check if message exceeded delivery timeout (80s)
            val nowMs = System.currentTimeMillis()
            val ageMs = nowMs - packet.timestamp
            if (ageMs >= OUTGOING_MAX_LIFETIME_MS) {
                addLog("⚠️ Message ${messageId.take(8)} expired after ${ageMs}ms — marking as error")
                markOutgoingAsError(messageId, packet)
                return@launch
            }

            if (attempts >= OUTGOING_MAX_RETRY_ATTEMPTS) {
                addLog("⚠️ Message ${messageId.take(8)} exhausted $attempts retries — marking as error")
                markOutgoingAsError(messageId, packet)
                return@launch
            }

            if (!isAuthenticated()) {
                // Not authenticated — defer to reconnect retry
                addLog("⏳ Message ${messageId.take(8)} retry deferred — not authenticated")
                resolveOutgoingRetry(messageId)
                return@launch
            }

            val nextAttempt = attempts + 1
            pendingOutgoingAttempts[messageId] = nextAttempt
            addLog("🔄 Retrying message ${messageId.take(8)}, attempt $nextAttempt")
            send(packet)
            scheduleOutgoingRetry(messageId)
        }
    }

    /**
     * iOS parity: mark an outgoing message as error and clean up retry state.
     */
    private fun markOutgoingAsError(messageId: String, packet: PacketMessage) {
        scope.launch {
            val repository = messageRepository ?: return@launch
            val opponentKey = if (packet.fromPublicKey == repository.getCurrentAccountKey())
                packet.toPublicKey else packet.fromPublicKey
            val dialogKey = repository.getDialogKey(opponentKey)
            repository.updateMessageDeliveryStatus(dialogKey, messageId, DeliveryStatus.ERROR)
        }
        resolveOutgoingRetry(messageId)
    }

    /**
     * iOS parity: cancel retry and clean up state for a resolved outgoing message.
     * Called when delivery ACK (0x08) is received.
     */
    fun resolveOutgoingRetry(messageId: String) {
        pendingOutgoingRetryJobs[messageId]?.cancel()
        pendingOutgoingRetryJobs.remove(messageId)
        pendingOutgoingPackets.remove(messageId)
        pendingOutgoingAttempts.remove(messageId)
    }

    /**
     * Cancel all pending outgoing retry jobs (e.g., on disconnect).
     */
    private fun cancelAllOutgoingRetries() {
        pendingOutgoingRetryJobs.values.forEach { it.cancel() }
        pendingOutgoingRetryJobs.clear()
        pendingOutgoingPackets.clear()
        pendingOutgoingAttempts.clear()
    }

    /**
     * Send packet (legacy name)
     */
    fun sendPacket(packet: Packet) {
        send(packet)
    }

    /**
     * Send call signaling packet (0x1A).
     */
    fun sendCallSignal(
        signalType: SignalType,
        src: String = "",
        dst: String = "",
        sharedPublic: String = "",
        roomId: String = ""
    ) {
        send(
            PacketSignalPeer().apply {
                this.signalType = signalType
                this.src = src
                this.dst = dst
                this.sharedPublic = sharedPublic
                this.roomId = roomId
            }
        )
    }

    /**
     * Send WebRTC signaling packet (0x1B).
     */
    fun sendWebRtcSignal(signalType: WebRTCSignalType, sdpOrCandidate: String) {
        send(
            PacketWebRTC().apply {
                this.signalType = signalType
                this.sdpOrCandidate = sdpOrCandidate
            }
        )
    }

    /**
     * Request ICE servers from server (0x1C).
     */
    fun requestIceServers() {
        send(PacketIceServers())
    }

    /**
     * Typed subscribe for call signaling packets (0x1A).
     * Returns wrapper callback for subsequent unwait.
     */
    fun waitCallSignal(callback: (PacketSignalPeer) -> Unit): (Packet) -> Unit {
        val wrapper: (Packet) -> Unit = { packet ->
            (packet as? PacketSignalPeer)?.let(callback)
        }
        waitPacket(PACKET_SIGNAL_PEER, wrapper)
        return wrapper
    }

    fun unwaitCallSignal(callback: (Packet) -> Unit) {
        unwaitPacket(PACKET_SIGNAL_PEER, callback)
    }

    /**
     * Typed subscribe for WebRTC packets (0x1B).
     * Returns wrapper callback for subsequent unwait.
     */
    fun waitWebRtcSignal(callback: (PacketWebRTC) -> Unit): (Packet) -> Unit {
        val wrapper: (Packet) -> Unit = { packet ->
            (packet as? PacketWebRTC)?.let(callback)
        }
        waitPacket(PACKET_WEB_RTC, wrapper)
        return wrapper
    }

    fun unwaitWebRtcSignal(callback: (Packet) -> Unit) {
        unwaitPacket(PACKET_WEB_RTC, callback)
    }

    /**
     * Typed subscribe for ICE servers packet (0x1C).
     * Returns wrapper callback for subsequent unwait.
     */
    fun waitIceServers(callback: (PacketIceServers) -> Unit): (Packet) -> Unit {
        val wrapper: (Packet) -> Unit = { packet ->
            (packet as? PacketIceServers)?.let(callback)
        }
        waitPacket(PACKET_ICE_SERVERS, wrapper)
        return wrapper
    }

    fun unwaitIceServers(callback: (Packet) -> Unit) {
        unwaitPacket(PACKET_ICE_SERVERS, callback)
    }
    
    /**
     * Register packet handler
     */
    fun waitPacket(packetId: Int, callback: (Packet) -> Unit) {
        getProtocol().waitPacket(packetId, callback)
    }
    
    /**
     * Unregister packet handler
     */
    fun unwaitPacket(packetId: Int, callback: (Packet) -> Unit) {
        getProtocol().unwaitPacket(packetId, callback)
    }

    private fun buildHandshakeDevice(): HandshakeDevice {
        val context = appContext
        val deviceId = if (context != null) {
            getOrCreateDeviceId(context)
        } else {
            generateDeviceId()
        }
        val manufacturer = Build.MANUFACTURER.orEmpty().trim()
        val model = Build.MODEL.orEmpty().trim()
        val name = listOf(manufacturer, model)
            .filter { it.isNotBlank() }
            .distinct()
            .joinToString(" ")
            .ifBlank { "Android Device" }
        val os = "Android ${Build.VERSION.RELEASE ?: "Unknown"}"

        return HandshakeDevice(
            deviceId = deviceId,
            deviceName = name,
            deviceOs = os
        )
    }

    private fun getOrCreateDeviceId(context: Context): String {
        val prefs = context.getSharedPreferences(DEVICE_PREFS, Context.MODE_PRIVATE)
        val cached = prefs.getString(DEVICE_ID_KEY, null)
        if (!cached.isNullOrBlank()) {
            return cached
        }
        val newId = generateDeviceId()
        prefs.edit().putString(DEVICE_ID_KEY, newId).apply()
        return newId
    }

    private fun generateDeviceId(): String {
        val chars = "abcdefghijklmnopqrstuvwxyz0123456789"
        val random = SecureRandom()
        return buildString(DEVICE_ID_LENGTH) {
            repeat(DEVICE_ID_LENGTH) {
                append(chars[random.nextInt(chars.length)])
            }
        }
    }
    
    /**
     * Disconnect and clear
     */
    fun disconnect() {
        protocol?.disconnect()
        protocol?.clearCredentials()
        messageRepository?.clearInitialization()
        clearTypingState()
        _devices.value = emptyList()
        _pendingDeviceVerification.value = null
        syncRequestInFlight = false
        clearSyncRequestTimeout()
        setSyncInProgress(false)
        resyncRequiredAfterAccountInit = false
        lastSubscribedToken = null  // reset so token is re-sent on next connect
    }
    
    /**
     * Destroy instance completely
     */
    fun destroy() {
        protocol?.destroy()
        protocol = null
        messageRepository?.clearInitialization()
        clearTypingState()
        _devices.value = emptyList()
        _pendingDeviceVerification.value = null
        syncRequestInFlight = false
        clearSyncRequestTimeout()
        setSyncInProgress(false)
        resyncRequiredAfterAccountInit = false
        scope.cancel()
    }
    
    /**
     * Check if authenticated
     */
    fun isAuthenticated(): Boolean = protocol?.isAuthenticated() ?: false
    
    /**
     * Check if connected
     */
    fun isConnected(): Boolean = protocol?.isConnected() ?: false
}
