package com.opendance.studio

import com.opendance.studio.net.PoseWebSocketClient
import com.opendance.studio.net.ReconnectPolicy
import com.opendance.studio.util.ConnectionLog
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import okhttp3.OkHttpClient
import okhttp3.Response
import okhttp3.WebSocket
import okhttp3.WebSocketListener
import okhttp3.mockwebserver.MockResponse
import okhttp3.mockwebserver.MockWebServer
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import org.junit.runner.RunWith
import org.robolectric.RobolectricTestRunner
import org.robolectric.annotation.Config
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger

/**
 * T-AND-016：**重连链路**的 JVM 集成测试（真实 WebSocket，不需要设备）
 *
 * 为什么单有 [com.opendance.studio.ReconnectPolicyTest] 不够：
 * `ReconnectPolicy` 是纯函数，测的是「给定关闭码应该怎么退避」；
 * 而 ⑦「频繁重连」的缺陷往往出在**调用点**——关闭码是否被正确记录、
 * 重连计数是否在成功后清零、上一次的关闭码会不会污染下一次会话。
 * 这些只有把**真实的 OkHttp WebSocket 接到一个真实的服务端**上才能验出来。
 *
 * 这里用 MockWebServer 的 `withWebSocketUpgrade` 起真服务端：
 *   * 能按契约 §3.2 回 `welcome`（含 silentTimeoutMs / heartbeatIntervalMs）；
 *   * 能按契约 §3.6 回 `pong`；
 *   * 能用**指定关闭码**关连接（1001 对端离开 / 1011 内部错误 / …），
 *     从而分别验证「不重连」与「按退避重连」两条分支。
 *
 * 注：Robolectric 只用来提供 `android.os.Build.*` 等桩（客户端构造时会读机型名）。
 */
@RunWith(RobolectricTestRunner::class)
@Config(sdk = [34])
class PoseWebSocketReconnectTest {

    private lateinit var server: MockWebServer
    private lateinit var scope: CoroutineScope
    private val clients = mutableListOf<PoseWebSocketClient>()

    /** 服务端侧收到过的文本消息（hello / ping / stats / bye） */
    private val serverTexts = CopyOnWriteArrayList<String>()
    /** 服务端侧见过的连接次数（含重连） */
    private val serverOpens = AtomicInteger(0)

    private var welcomeSilentTimeoutMs = 4_500L
    private var closeCodeOnHello = 0
    private var closeAfterHelloDelayMs = 0L

    @Before
    fun setUp() {
        ConnectionLog.clear()
        serverTexts.clear()
        serverOpens.set(0)
        server = MockWebServer()
        server.start()
    }

    @After
    fun tearDown() {
        clients.forEach { runCatching { it.disconnect() } }
        scope.cancel()
        runCatching { server.shutdown() }
        ConnectionLog.clear()
    }

    // ------------------------------------------------------------------ 夹具

    private fun enqueueWebSocket() {
        server.enqueue(
            // 必须回显子协议：RFC 6455 要求服务端在 101 响应里带上选中的
            // Sec-WebSocket-Protocol，客户端据此记录「协商结果」。
            // （第一版夹具没带，客户端 negotiatedProtocol 就是空串 —— 那是**夹具**不合规，
            //   不是客户端的问题；这条断言因此改成先在夹具里把服务端做对。）
            MockResponse()
                .setHeader("Sec-WebSocket-Protocol", "opendance.pose.v1")
                .withWebSocketUpgrade(object : WebSocketListener() {
                override fun onOpen(webSocket: WebSocket, response: Response) {
                    serverOpens.incrementAndGet()
                }

                override fun onMessage(webSocket: WebSocket, text: String) {
                    serverTexts.add(text)
                    when {
                        text.contains("\"hello\"") -> {
                            webSocket.send(
                                """{"type":"welcome","protocol":"opendance.pose.v1",""" +
                                    """"sessionId":"sess-$serverOpens","expectedFps":30,""" +
                                    """"silentTimeoutMs":$welcomeSilentTimeoutMs,""" +
                                    """"heartbeatIntervalMs":${ReconnectPolicy.heartbeatIntervalMs(welcomeSilentTimeoutMs)}}""",
                            )
                            if (closeCodeOnHello != 0) {
                                // 模拟「对端主动关闭 / 服务重启」等场景
                                Thread {
                                    Thread.sleep(closeAfterHelloDelayMs)
                                    webSocket.close(closeCodeOnHello, "test-close-$closeCodeOnHello")
                                }.start()
                            }
                        }
                        text.contains("\"ping\"") -> {
                            webSocket.send("""{"type":"pong","sentAtMs":1,"serverAtMs":2}""")
                        }
                    }
                }
            }),
        )
    }

    private fun newClient(): PoseWebSocketClient {
        scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
        return PoseWebSocketClient(scope).also { clients.add(it) }
    }

    private fun wsUrl(): String = server.url("/pose").toString().replace("http://", "ws://")

    /** 轮询等待条件成立（避免依赖固定 sleep 造成的偶发失败） */
    private fun waitFor(timeoutMs: Long = 6_000, what: String, cond: () -> Boolean): Boolean {
        val deadline = System.currentTimeMillis() + timeoutMs
        while (System.currentTimeMillis() < deadline) {
            if (cond()) return true
            Thread.sleep(25)
        }
        throw AssertionError("等待超时（${timeoutMs}ms）：$what")
    }

    private fun waitUntilOrFalse(timeoutMs: Long, cond: () -> Boolean): Boolean {
        val deadline = System.currentTimeMillis() + timeoutMs
        while (System.currentTimeMillis() < deadline) {
            if (cond()) return true
            Thread.sleep(25)
        }
        return false
    }

    // ------------------------------------------------------------------ 用例

    @Test
    fun `握手成功后进入 streaming 并解析 welcome 的保活参数`() {
        enqueueWebSocket()
        val client = newClient()
        client.connect(wsUrl())

        waitFor(what = "进入 streaming") { client.stats.value.state == PoseWebSocketClient.State.STREAMING }

        val stats = client.stats.value
        assertTrue("sessionId 应来自 welcome（实际 '${stats.serverSessionId}'）", stats.serverSessionId.isNotEmpty())
        assertEquals("opendance.pose.v1", stats.negotiatedProtocol)
        assertEquals(welcomeSilentTimeoutMs, stats.silentTimeoutMs)
        assertEquals(ReconnectPolicy.heartbeatIntervalMs(welcomeSilentTimeoutMs), stats.heartbeatIntervalMs)
        assertTrue(
            "hello 必须带 deviceId（契约 §3.1，接收端据此去重）",
            serverTexts.any { it.contains("\"type\":\"hello\"") && it.contains("\"deviceId\"") },
        )
    }

    @Test
    fun `按协商的心跳间隔发送应用层 ping 并收到 pong`() {
        enqueueWebSocket()
        val client = newClient()
        client.connect(wsUrl())
        waitFor(what = "进入 streaming") { client.stats.value.state == PoseWebSocketClient.State.STREAMING }

        waitFor(timeoutMs = 8_000, what = "收到 pong") { client.stats.value.pongsReceived >= 1 }
        assertTrue(
            "服务端必须真的收到 ping 文本帧（协议层 ping 对应用层不可见，不能替代）",
            serverTexts.any { it.contains("\"type\":\"ping\"") },
        )
    }

    @Test
    fun `异常关闭码（1011）会按退避策略重连且成功后计数清零`() {
        closeCodeOnHello = 1011
        enqueueWebSocket()   // 第一次连接：握手后立刻被服务端以 1011 关闭
        enqueueWebSocket()   // 重连后的第二次连接
        val client = newClient()
        client.connect(wsUrl())

        // 第一次连接被关闭 → 进入 reconnecting 且尝试次数为 1，等待时间落在策略区间内
        waitFor(what = "进入 reconnecting") { client.stats.value.state == PoseWebSocketClient.State.RECONNECTING }
        val firstAttempt = client.stats.value.reconnectAttempts
        assertEquals(1, firstAttempt)
        val expected = ReconnectPolicy.baseDelayMs(1)                       // 500ms
        val low = (expected * (1 - ReconnectPolicy.JITTER_RATIO)).toLong()   // 400ms
        val high = (expected * (1 + ReconnectPolicy.JITTER_RATIO)).toLong()  // 600ms
        assertTrue(
            "第 1 次重连等待应落在策略区间 [$low, $high]（实际 ${client.stats.value.nextRetryMs}）",
            client.stats.value.nextRetryMs in low..high,
        )

        // 真的重连上了（服务端看到第 2 次 upgrade），并且计数清零、状态回到 streaming
        waitFor(timeoutMs = 8_000, what = "第二次连接建立") { serverOpens.get() >= 2 }
        waitFor(what = "重连后回到 streaming") { client.stats.value.state == PoseWebSocketClient.State.STREAMING }
        assertEquals("连接成功后重连计数必须清零（契约 §7.1）", 0, client.stats.value.reconnectAttempts)
    }

    @Test
    fun `1001 对端离开不重连`() {
        closeCodeOnHello = 1001
        enqueueWebSocket()
        enqueueWebSocket()
        val client = newClient()
        client.connect(wsUrl())

        waitFor(what = "进入 stopped") { client.stats.value.state == PoseWebSocketClient.State.STOPPED }
        assertEquals(1001, client.stats.value.lastCloseCode)
        assertTrue("1001 不应重连（契约 §7）", client.stats.value.reconnectAttempts == 0)
        val reconnected = waitUntilOrFalse(2_000) { serverOpens.get() >= 2 }
        assertTrue("1001 之后不允许再建立新连接", !reconnected)
        assertTrue(
            "日志里应记录「不重连」便于真机排障",
            ConnectionLog.entries().any { it.contains("不重连") },
        )
    }

    @Test
    fun `1002 协议错误不重连`() {
        closeCodeOnHello = 1002
        enqueueWebSocket()
        enqueueWebSocket()
        val client = newClient()
        client.connect(wsUrl())

        waitFor(what = "进入 stopped") { client.stats.value.state == PoseWebSocketClient.State.STOPPED }
        assertEquals(1002, client.stats.value.lastCloseCode)
        assertTrue("1002 不应重连（重连也不会成功）", !waitUntilOrFalse(2_000) { serverOpens.get() >= 2 })
    }

    @Test
    fun `用户主动 disconnect 不重连`() {
        enqueueWebSocket()
        enqueueWebSocket()
        val client = newClient()
        client.connect(wsUrl())
        waitFor(what = "进入 streaming") { client.stats.value.state == PoseWebSocketClient.State.STREAMING }

        client.disconnect()
        waitFor(what = "进入 stopped") { client.stats.value.state == PoseWebSocketClient.State.STOPPED }
        assertTrue("主动断开不应重连", !waitUntilOrFalse(2_000) { serverOpens.get() >= 2 })
    }

    /**
     * 本用例是 ⑦ 的关键回归：**上一次会话的关闭码不能污染下一次会话**。
     *
     * 复现路径（修复前）：
     *   1. 第一次连接被对端以 1001 关闭 → lastCloseCode = 1001，按契约不重连（正确）；
     *   2. 用户再次点「开始推流」→ connect()；
     *   3. 这一次是**网络层失败**（服务端还没起来 / 被拒），应当按 §7 重连；
     *   4. 但 `connect()` 没有清理 lastCloseCode，scheduleReconnect() 仍按 1001 判定
     *      → 直接 stopped，**再也不会自动重连**，用户只能反复手点。
     */
    @Test
    fun `上一次 1001 关闭后重新连接，遇到网络失败仍应按策略重连`() {
        // 1. 第一次：对端以 1001 关闭
        closeCodeOnHello = 1001
        enqueueWebSocket()
        val client = newClient()
        client.connect(wsUrl())
        waitFor(what = "第一次连接进入 stopped") { client.stats.value.state == PoseWebSocketClient.State.STOPPED }
        assertEquals(1001, client.stats.value.lastCloseCode)

        // 2. 用户重新开始推流，但这一轮服务端 connect 就会失败（直接关掉 MockWebServer）
        closeCodeOnHello = 0
        server.shutdown()
        client.connect(server.url("/pose").toString().replace("http://", "ws://"))

        // 3. 网络失败必须触发重连（进入 reconnecting 或至少 attempts > 0）
        val recovered = waitUntilOrFalse(6_000) {
            client.stats.value.state == PoseWebSocketClient.State.RECONNECTING ||
                client.stats.value.reconnectAttempts > 0
        }
        assertTrue(
            "上一次的 1001 不应影响本次会话：网络失败后必须按 §7 重连" +
                "（当前 state=${client.stats.value.state}，" +
                "lastCloseCode=${client.stats.value.lastCloseCode}，" +
                "attempts=${client.stats.value.reconnectAttempts}）",
            recovered,
        )
    }

    @Test
    fun `服务端未启动时的连续重连尝试遵守退避（不会疯狂重连）`() {
        server.shutdown()   // 没有服务端
        val client = newClient()
        val t0 = System.currentTimeMillis()
        client.connect(server.url("/pose").toString().replace("http://", "ws://"))

        waitFor(timeoutMs = 6_000, what = "至少安排一次重连") { client.stats.value.reconnectAttempts >= 1 }
        // 1.5 秒内不应超过 3 次尝试：0.5s → 1s → 2s，说明退避生效
        Thread.sleep(1_500)
        val attempts = client.stats.value.reconnectAttempts
        val elapsed = System.currentTimeMillis() - t0
        assertTrue(
            "退避必须生效：${elapsed}ms 内只应有少量尝试（实际 $attempts）",
            attempts <= 3,
        )
        assertNotNull(client.stats.value.lastError)
    }
}