package com.tokilabs.proto_socket

import com.tokilabs.proto_socket.packets.HeartBeat
import com.tokilabs.proto_socket.packets.PacketBase
import com.tokilabs.proto_socket.packets.TestData
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.async
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.test.runTest
import kotlinx.coroutines.withContext
import java.util.Collections
import java.util.concurrent.atomic.AtomicInteger
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFailsWith
import kotlin.test.assertTrue

private class FakeTransport : Transport {
    val packets: MutableList<PacketBase> = Collections.synchronizedList(mutableListOf())

    override suspend fun writePacket(base: PacketBase) {
        packets.add(base)
    }

    override fun close() {
    }
}

private fun Communicator.setNonceForTest(value: Int) {
    val field = Communicator::class.java.getDeclaredField("nonce")
    field.isAccessible = true
    (field.get(this) as AtomicInteger).set(value)
}

class CommunicatorTest {
    @Test
    fun testSendRequestTimeout() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        val err = assertFailsWith<IllegalStateException> {
            sendRequestTyped<TestData, TestData>(
                communicator,
                TestData.newBuilder().setIndex(1).setMessage("no response").build(),
                timeoutMs = 25,
            )
        }
        assertTrue(err.message!!.contains("timeout"))
        assertEquals(1, transport.packets.size)
        communicator.close()
    }

    @Test
    fun testSendRequestTypeMismatch() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        val result = async {
            assertFailsWith<IllegalStateException> {
                sendRequestTyped<TestData, TestData>(
                    communicator,
                    TestData.newBuilder().setIndex(1).setMessage("hello").build(),
                    timeoutMs = 1_000,
                )
            }
        }
        while (transport.packets.isEmpty()) delay(1)
        val requestNonce = transport.packets.single().nonce
        communicator.onReceivedData(
            typeNameOf<HeartBeat>(),
            HeartBeat.getDefaultInstance().toByteArray(),
            responseNonce = requestNonce,
        )
        assertTrue(result.await().message!!.contains("response type mismatch"))
        communicator.close()
    }

    @Test
    fun testListenerAndRequestListenerConflict() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        addListenerTyped<TestData>(communicator) {}
        assertFailsWith<IllegalStateException> {
            addRequestListenerTyped<TestData, TestData>(communicator) { it }
        }
        communicator.close()
    }

    @Test
    fun testSendFireAndForget() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        communicator.send(TestData.newBuilder().setIndex(7).setMessage("fire").build())

        assertEquals(1, transport.packets.size)
        assertEquals(typeNameOf<TestData>(), transport.packets.single().typeName)
        communicator.close()
    }

    @Test
    fun testNonceWrapsAfterIntMaxWithoutEmittingZero() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)
        communicator.setNonceForTest(Int.MAX_VALUE - 1)

        communicator.send(TestData.newBuilder().setIndex(1).build())
        communicator.send(TestData.newBuilder().setIndex(2).build())

        assertEquals(listOf(Int.MAX_VALUE, 1), transport.packets.map { it.nonce })
        assertTrue(transport.packets.none { it.nonce == 0 })
        communicator.close()
    }

    @Test
    fun testSendRequestNonceWrapsAtInt32Max() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)
        communicator.setNonceForTest(Int.MAX_VALUE - 1)

        val result = async {
            sendRequestTyped<TestData, TestData>(
                communicator,
                TestData.newBuilder().setIndex(1).setMessage("max").build(),
                timeoutMs = 1_000,
            )
        }
        while (transport.packets.isEmpty()) delay(1)
        val request = transport.packets.single()
        assertEquals(Int.MAX_VALUE, request.nonce)
        assertTrue(request.nonce != 0)

        val response = TestData.newBuilder().setIndex(2).setMessage("max response").build()
        communicator.onReceivedData(
            typeNameOf<TestData>(),
            response.toByteArray(),
            responseNonce = request.nonce,
        )

        assertEquals("max response", result.await().message)
        communicator.close()
    }

    @Test
    fun testNextNonceSkipsActivePendingAfterWrap() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)
        communicator.setNonceForTest(Int.MAX_VALUE)

        val result = async {
            sendRequestTyped<TestData, TestData>(
                communicator,
                TestData.newBuilder().setIndex(1).setMessage("pending-one").build(),
                timeoutMs = 10_000,
            )
        }
        while (transport.packets.isEmpty()) delay(1)
        val firstRequest = transport.packets[0]
        assertEquals(1, firstRequest.nonce)

        communicator.send(TestData.newBuilder().setIndex(2).setMessage("skip-one").build())
        while (transport.packets.size < 2) delay(1)
        val secondPacket = transport.packets[1]
        assertEquals(2, secondPacket.nonce)

        val response = TestData.newBuilder().setIndex(3).setMessage("pending-one response").build()
        communicator.onReceivedData(
            typeNameOf<TestData>(),
            response.toByteArray(),
            responseNonce = 1,
        )
        result.await()
        communicator.close()
    }

    @Test
    fun testInboundQueueFifoOrder() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        val received = mutableListOf<Int>()
        addListenerTyped<TestData>(communicator) { msg -> received.add(msg.index) }

        for (i in 1..5) {
            val data = TestData.newBuilder().setIndex(i).build().toByteArray()
            communicator.onReceivedData(typeNameOf<TestData>(), data, incomingNonce = i)
        }

        // receiveLoop가 처리하도록 yield
        var elapsed = 0
        while (received.size < 5 && elapsed < 1000) {
            delay(5)
            elapsed += 5
        }

        assertEquals(listOf(1, 2, 3, 4, 5), received, "FIFO violated")
        communicator.close()
    }

    @Test
    fun testSlowRequestHandlerResponseFifo() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        addRequestListenerTyped<TestData, TestData>(communicator) { req ->
            if (req.index == 1) delay(50) // 첫 번째 요청 50ms 지연
            TestData.newBuilder().setIndex(req.index + 100).setMessage("echo").build()
        }

        val data1 = TestData.newBuilder().setIndex(1).setMessage("first").build().toByteArray()
        val data2 = TestData.newBuilder().setIndex(2).setMessage("second").build().toByteArray()
        communicator.onReceivedData(typeNameOf<TestData>(), data1, incomingNonce = 10)
        communicator.onReceivedData(typeNameOf<TestData>(), data2, incomingNonce = 20)

        var elapsed = 0
        while (transport.packets.size < 2 && elapsed < 2000) {
            delay(10)
            elapsed += 10
        }

        assertTrue(transport.packets.size >= 2, "expected 2 responses, got ${transport.packets.size}")
        // FIFO: 첫 번째 응답이 responseNonce=10, 두 번째가 responseNonce=20
        assertEquals(10, transport.packets[0].responseNonce, "first response should be for nonce 10")
        assertEquals(20, transport.packets[1].responseNonce, "second response should be for nonce 20")
        communicator.close()
    }

    @Test
    fun testCloseCancelsPendingRequests() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        val result = async {
            assertFailsWith<NotConnectedException> {
                sendRequestTyped<TestData, TestData>(
                    communicator,
                    TestData.newBuilder().setIndex(99).setMessage("pending").build(),
                    timeoutMs = 5_000,
                )
            }
        }

        while (transport.packets.isEmpty()) delay(1)
        communicator.close()
        result.await() // NotConnectedException이 throw 되어야 함
    }

    @Test
    fun testInboundBackpressureOnHeavyLoad() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        val deferred = kotlinx.coroutines.CompletableDeferred<Unit>()

        // 지연 request 핸들러 등록
        communicator.addRequestListener(typeNameOf<TestData>()) { req, nonce ->
            deferred.await()
            TestData.newBuilder().setIndex(100).build()
        }

        val data = TestData.newBuilder().setIndex(1).setMessage("heavy").build().toByteArray()

        // 1번째 호출: 핸들러가 deferred.await()에 의해 블로킹됨
        communicator.enqueueInbound(typeNameOf<TestData>(), data, incomingNonce = 1)

        // 2번째부터 65번째(총 64개)를 enqueue: 첫번째 아이템은 디스패치 중이므로 큐에는 현재 0개 있음
        // 2 ~ 65번째(총 64개) 추가 시 큐가 꽉 차게 됨
        for (i in 2..65) {
            communicator.enqueueInbound(typeNameOf<TestData>(), data, incomingNonce = i)
        }

        // 66번째 item 추가 시도: suspend(대기) 상태여야 함
        val enqueueJob = async {
            communicator.enqueueInbound(typeNameOf<TestData>(), data, incomingNonce = 66)
        }

        delay(50)
        assertTrue(!enqueueJob.isCompleted, "enqueueInbound should suspend when the queue is full")

        // 핸들러를 완료시켜 큐를 하나 소모하게 함
        deferred.complete(Unit)

        // 대기 중이던 enqueueJob이 무사히 재개되어 완료되어야 함
        enqueueJob.await()
        communicator.shutdown()
    }

    @Test
    fun testGracefulCloseDrainsInFlight() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        var handlerStarted = false
        var handlerFinished = false

        addRequestListenerTyped<TestData, TestData>(communicator) { req ->
            handlerStarted = true
            delay(100)
            handlerFinished = true
            TestData.newBuilder().setIndex(100).setMessage("graceful").build()
        }

        val data = TestData.newBuilder().setIndex(1).build().toByteArray()
        communicator.onReceivedData(typeNameOf<TestData>(), data, incomingNonce = 1)

        // 핸들러가 구동을 개시할 때까지 대기
        var elapsed = 0
        while (!handlerStarted && elapsed < 1000) {
            delay(5)
            elapsed += 5
        }
        assertTrue(handlerStarted, "handler should start")

        // 이 상태에서 close() 호출
        communicator.close()

        // close() 가 완료된 시점에 핸들러가 무사히 완주하여 finishes 되었는지 확인
        assertTrue(handlerFinished, "in-flight request handler should finish execution before close completes")

        // 결과 검증: 자동 응답 패킷이 전송 큐에 정상 적재되었는지 확인
        assertEquals(1, transport.packets.size)
        assertEquals(1, transport.packets.single().responseNonce)
        assertTrue(!communicator.isAlive())
    }

    @Test
    fun testWsInboundStagingOverflowDisconnects() = runTest {
        val connection = object : WsConnection {
            var closed = false
            override fun send(bytes: ByteArray): Boolean = true
            override fun close() {
                closed = true
            }
        }
        val client = WsClient(
            connection,
            intervalSec = 30,
            waitSec = 10,
            parserMap = testParserMap(),
            maxStagedFrames = 4,
        )
        
        val deferred = kotlinx.coroutines.CompletableDeferred<Unit>()
        client.communicator.addRequestListener(typeNameOf<TestData>()) { req, nonce ->
            deferred.await()
            TestData.newBuilder().setIndex(100).build()
        }

        val data = TestData.newBuilder().setIndex(1).setMessage("heavy").build().toByteArray()
        val frame = PacketBase.newBuilder()
            .setTypeName(typeNameOf<TestData>())
            .setNonce(1)
            .setData(com.google.protobuf.ByteString.copyFrom(data))
            .build()
            .toByteArray()

        client.receiveBytes(frame)

        for (i in 2..20) {
            val f = PacketBase.newBuilder()
                .setTypeName(typeNameOf<TestData>())
                .setNonce(i)
                .setData(com.google.protobuf.ByteString.copyFrom(data))
                .build()
                .toByteArray()
            client.receiveBytes(f)
        }

        // staging 코루틴은 Dispatchers.IO(실제 스레드)에서 동작하므로 runTest 가상 시간 delay로는
        // onDisconnected의 실제 close 완료를 기다릴 수 없다. 실제 시간 기준으로 connection.closed를 폴링한다.
        withContext(Dispatchers.IO) {
            val deadline = System.nanoTime() + 2_000_000_000L
            while (System.nanoTime() < deadline && !connection.closed) {
                Thread.sleep(5)
            }
        }

        assertTrue(connection.closed, "connection should be closed forcefully due to staging overflow")
        assertTrue(!client.communicator.isAlive(), "communicator should be inactive")

        val snapshot = client.getMetricsSnapshot()
        assertTrue(snapshot.maxStaged > 0, "maxStaged should be > 0, got ${snapshot.maxStaged}")
        assertTrue(snapshot.droppedCount > 0, "droppedCount should be > 0, got ${snapshot.droppedCount}")
        assertTrue(snapshot.receivedCount >= 5L, "receivedCount should be >= 5, got ${snapshot.receivedCount}")
        assertEquals(snapshot.receivedCount, snapshot.enqueuedCount + snapshot.droppedCount, "receivedCount = enqueued + dropped")

        deferred.complete(Unit)
    }

    @Test
    fun testTcpInboundBackpressureFlowControl() = runTest {
        val serverSocket = java.net.ServerSocket(0)
        val port = serverSocket.localPort

        val deferred = kotlinx.coroutines.CompletableDeferred<Unit>()
        var serverClient: TcpClient? = null

        val serverJob = this.launch(Dispatchers.IO) {
            val socket = serverSocket.accept()
            serverClient = TcpClient.fromSocket(socket, intervalSec = 30, waitSec = 10, parserMap = testParserMap())
            serverClient!!.communicator.addRequestListener(typeNameOf<TestData>()) { req, nonce ->
                runBlocking { deferred.await() }
                TestData.newBuilder().setIndex(100).build()
            }
        }

        val client = dialTcp("127.0.0.1", port, intervalSec = 30, waitSec = 10, parserMap = testParserMap())

        client.send(TestData.newBuilder().setIndex(1).setMessage("first").build())
        delay(50)

        val heavy = TestData.newBuilder().setIndex(2).setMessage("heavy").build()
        for (i in 2..65) {
            client.send(heavy)
        }

        client.send(TestData.newBuilder().setIndex(66).setMessage("overflow").build())
        delay(100)

        // enqueueInbound가 블록되었을 때 readLoop 코루틴이 여전히 동작 중이지만 완료되지 않았는지 확인
        // serverClient 내부 readLoopJob이 활성 상태(isActive)이고 완료(isCompleted)되지 않은 상태로 대기하고 있어야 함
        val readLoopJob = try {
            val sClient = serverClient
            if (sClient != null) {
                val field = sClient.javaClass.getDeclaredField("communicator")
                field.isAccessible = true
                val comm = field.get(sClient)
                val loopField = comm.javaClass.getDeclaredField("receiveLoopJob")
                loopField.isAccessible = true
                loopField.get(comm) as? kotlinx.coroutines.Job
            } else {
                null
            }
        } catch (e: Exception) {
            null
        }

        if (readLoopJob != null) {
            assertTrue(readLoopJob.isActive, "readLoop should be active but blocked")
            assertTrue(!readLoopJob.isCompleted, "readLoop should not be completed")
        }

        deferred.complete(Unit)
        delay(50)

        client.close()
        serverSocket.close()
        serverJob.cancel()
    }

    @Test
    fun testWorkerGatewayReorderAndDispatch() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        val received = mutableListOf<Int>()
        addListenerTyped<TestData>(communicator) { msg -> received.add(msg.index) }

        // Fake gateway: out-of-order(seq=2, seq=1)로 전달된 아이템
        data class FakeGatewayItem(val seq: Int, val index: Int, val incomingNonce: Int)
        val items = mutableListOf(
            FakeGatewayItem(seq = 2, index = 2, incomingNonce = 2),
            FakeGatewayItem(seq = 1, index = 1, incomingNonce = 1),
        )

        // coordinator로 밀어넣기 전에 seq 오름차순으로 reorder
        items.sortBy { it.seq }

        for (item in items) {
            val data = TestData.newBuilder().setIndex(item.index).build().toByteArray()
            communicator.onReceivedData(typeNameOf<TestData>(), data, incomingNonce = item.incomingNonce)
        }

        // receiveLoop가 처리하도록 yield
        var elapsed = 0
        while (received.size < 2 && elapsed < 1000) {
            delay(5)
            elapsed += 5
        }

        assertEquals(listOf(1, 2), received, "seq ordering was not preserved")
        communicator.close()
    }

    @Test
    fun testFrameReorderBufferReleasesInOrder() {
        val buffer = FrameReorderBuffer()

        assertTrue(
            buffer.release(GatewayResult(seq = 3, frame = null, error = null)).isEmpty(),
            "seq 3 should buffer until 1,2 arrive",
        )
        assertTrue(
            buffer.release(GatewayResult(seq = 2, frame = null, error = null)).isEmpty(),
            "seq 2 should still buffer until 1 arrives",
        )
        val flushed = buffer.release(GatewayResult(seq = 1, frame = null, error = null))
        assertEquals(listOf(1L, 2L, 3L), flushed.map { it.seq }, "1,2,3 should flush together in order")

        // nextSeq 아래 seq는 중복/지연 도착이므로 무시되어야 한다.
        assertTrue(
            buffer.release(GatewayResult(seq = 1, frame = null, error = null)).isEmpty(),
            "late duplicate seq should be ignored",
        )
    }

    @Test
    fun testWorkerGatewayReordersOutOfOrderCompletion() = runTest {
        val total = 8
        val dispatched = mutableListOf<Int>()

        // 낮은 nonce(앞선 seq)일수록 더 오래 delay시켜 worker 완료 순서를 역전시킨다.
        val gateway = CoroutineWorkerGateway(
            scope = this,
            workers = 4,
            queueSize = 16,
            decode = { bytes ->
                val base = PacketBase.parseFrom(bytes)
                delay(((total - base.nonce) * 5).toLong())
                base
            },
            sink = { frame -> dispatched.add(frame.incomingNonce) },
        )

        for (seq in 1..total) {
            val frame = PacketBase.newBuilder().setTypeName("x").setNonce(seq).build().toByteArray()
            gateway.submit(InboundFrame(seq = seq.toLong(), bytes = frame))
        }

        var elapsed = 0
        while (dispatched.size < total && elapsed < 5000) {
            delay(5)
            elapsed += 5
        }

        assertEquals((1..total).toList(), dispatched, "seq ordering not preserved despite out-of-order completion")
        gateway.close()
    }

    @Test
    fun testWorkerGatewayReportsDecodeErrorAndAdvances() = runTest {
        val seen = mutableListOf<Int>()
        var errorCount = 0

        val gateway = CoroutineWorkerGateway(
            scope = this,
            workers = 2,
            queueSize = 8,
            sink = { frame -> seen.add(frame.incomingNonce) },
            onError = { errorCount++ },
        )

        val good1 = PacketBase.newBuilder().setTypeName("x").setNonce(1).build().toByteArray()
        val good2 = PacketBase.newBuilder().setTypeName("x").setNonce(3).build().toByteArray()

        gateway.submit(InboundFrame(seq = 1, bytes = good1))
        gateway.submit(InboundFrame(seq = 2, bytes = byteArrayOf(0xff.toByte(), 0xff.toByte(), 0xff.toByte())))
        gateway.submit(InboundFrame(seq = 3, bytes = good2))

        var elapsed = 0
        while (seen.size < 2 && elapsed < 2000) {
            delay(5)
            elapsed += 5
        }

        assertEquals(listOf(1, 3), seen, "sink should see good frames in order, skipping the bad one")
        assertEquals(1, errorCount, "exactly one decode error should be reported")
        gateway.close()
    }

    @Test
    fun testAttachedGatewayDispatchesThroughOnReceivedFrameInOrder() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        val received = mutableListOf<Int>()
        addListenerTyped<TestData>(communicator) { msg -> received.add(msg.index) }

        val total = 50
        // production attached path: gateway를 communicator에 attach하고 raw frame을 onReceivedFrame으로
        // ingest한다. gateway sink는 decoded frame을 coordinator inbound queue로만 전달한다.
        val gateway = CoroutineWorkerGateway(
            scope = this,
            workers = 4,
            queueSize = 64,
            decode = { bytes ->
                val base = PacketBase.parseFrom(bytes)
                // 짝수 nonce를 더 오래 delay시켜 worker 완료 순서를 섞는다.
                delay(if (base.nonce % 2 == 0) 8L else 2L)
                base
            },
            sink = { frame ->
                communicator.enqueueInbound(frame.typeName, frame.data, frame.incomingNonce, frame.responseNonce)
            },
        )
        communicator.attachInboundGateway(gateway)

        for (i in 1..total) {
            val data = TestData.newBuilder().setIndex(i).build().toByteArray()
            val raw = PacketBase.newBuilder()
                .setTypeName(typeNameOf<TestData>())
                .setNonce(i)
                .setData(com.google.protobuf.ByteString.copyFrom(data))
                .build()
                .toByteArray()
            communicator.onReceivedFrame(raw)
        }

        var elapsed = 0
        while (received.size < total && elapsed < 5000) {
            delay(5)
            elapsed += 5
        }

        assertEquals((1..total).toList(), received, "coordinator dispatch order should match input order")
        communicator.close()
    }

    @Test
    fun testWsIngressSubmitsRawFramesToAttachedGateway() = runTest {
        val connection = object : WsConnection {
            override fun send(bytes: ByteArray): Boolean = true
            override fun close() {}
        }
        val client = WsClient(connection, intervalSec = 30, waitSec = 10, parserMap = testParserMap())

        // counting gateway: raw frame submit만 기록하고 decode하지 않는다.
        // read loop가 gateway를 우회해 inline decode하면 submitted가 비어 테스트가 실패한다.
        val submitted = Collections.synchronizedList(mutableListOf<Long>())
        val gateway = object : InboundGateway {
            override suspend fun submit(frame: InboundFrame) {
                submitted.add(frame.seq)
            }

            override fun close() {}
        }
        client.communicator.attachInboundGateway(gateway)

        val total = 5
        for (i in 1..total) {
            val data = TestData.newBuilder().setIndex(i).build().toByteArray()
            val frame = PacketBase.newBuilder()
                .setTypeName(typeNameOf<TestData>())
                .setNonce(i)
                .setData(com.google.protobuf.ByteString.copyFrom(data))
                .build()
                .toByteArray()
            client.receiveBytes(frame)
        }

        withContext(Dispatchers.IO) {
            val deadline = System.nanoTime() + 2_000_000_000L
            while (System.nanoTime() < deadline && submitted.size < total) {
                Thread.sleep(5)
            }
        }

        // ingress가 gateway에 frame을 submit했고 내부 seq가 수신 순서대로 부여되었는지 확인한다.
        assertEquals((1L..total.toLong()).toList(), submitted.sorted(), "WS ingress should submit raw frames to the attached gateway")
        client.communicator.shutdown()
    }

    @Test
    fun testWsGatewayDecodeErrorTriggersDisconnect() = runTest {
        val connection = object : WsConnection {
            var closed = false
            override fun send(bytes: ByteArray): Boolean = true
            override fun close() {
                closed = true
            }
        }
        val client = WsClient(connection, intervalSec = 30, waitSec = 10, parserMap = testParserMap())

        // production worker-pool gateway를 enable한다. decode 불가 frame은 gateway onError →
        // frameErrorHandler(WsClient가 onDisconnected로 등록) 경로로 transport disconnect를 유발해야 한다.
        client.communicator.enableInboundGateway()

        client.receiveBytes(byteArrayOf(0xff.toByte(), 0xff.toByte(), 0xff.toByte()))

        withContext(Dispatchers.IO) {
            val deadline = System.nanoTime() + 2_000_000_000L
            while (System.nanoTime() < deadline && !connection.closed) {
                Thread.sleep(5)
            }
        }

        assertTrue(connection.closed, "gateway decode error should reach the transport disconnect path")
        assertTrue(!client.communicator.isAlive(), "communicator should be inactive after gateway decode error")
    }

    @Test
    fun testWorkerGatewayCloseIsIdempotentAndStopsDelivery() = runTest {
        val dispatched = mutableListOf<Int>()

        val gateway = CoroutineWorkerGateway(
            scope = this,
            workers = 2,
            queueSize = 8,
            sink = { frame -> dispatched.add(frame.incomingNonce) },
        )

        gateway.close()
        gateway.close() // 멱등: 두 번 호출해도 안전해야 한다.

        // close 이후 submit은 drop되어 sink에 도달하지 않아야 한다.
        val frame = PacketBase.newBuilder().setTypeName("x").setNonce(1).build().toByteArray()
        gateway.submit(InboundFrame(seq = 1, bytes = frame))

        delay(50)
        assertTrue(dispatched.isEmpty(), "submit after close should be dropped")
        gateway.close()
    }

    @Test
    fun testAliasCollisionOnInit() = runTest {
        val badParserMap: ParserMap = mapOf(
            "a.b.TestData" to { TestData.parseFrom(it) },
            "x.y.TestData" to { TestData.parseFrom(it) },
        )
        val transport = FakeTransport()
        assertFailsWith<IllegalArgumentException> {
            Communicator(transport, badParserMap, this)
        }
    }

    @Test
    fun testLegacyAliasRouting() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        // 1. Test Listener routing with legacy simple name
        var listenerCalled = false
        var receivedMsg: com.google.protobuf.MessageLite? = null
        addListenerTyped<TestData>(communicator) { msg ->
            listenerCalled = true
            receivedMsg = msg
        }

        val testMsgBytes = TestData.newBuilder().setIndex(42).setMessage("test").build().toByteArray()
        communicator.onReceivedData("TestData", testMsgBytes, incomingNonce = 1)

        var elapsed = 0
        while (!listenerCalled && elapsed < 1000) {
            delay(5)
            elapsed += 5
        }
        assertTrue(listenerCalled, "listener should be called with legacy simple name")
        assertEquals(42, (receivedMsg as TestData).index)

        // 2. Test Pending request matching with legacy simple name
        val result = async {
            sendRequestTyped<TestData, TestData>(
                communicator,
                TestData.newBuilder().setIndex(1).setMessage("req").build(),
                timeoutMs = 1_000,
            )
        }

        while (transport.packets.isEmpty()) delay(1)
        val requestNonce = transport.packets.single().nonce

        communicator.onReceivedData("TestData", testMsgBytes, responseNonce = requestNonce)

        val response = result.await()
        assertEquals(42, response.index)
        communicator.close()
    }

    @Test
    fun testRequestHandlerAliasSimpleNameInbound() = runTest {
        val transport = FakeTransport()
        val communicator = Communicator(transport, testParserMap(), this)

        addRequestListenerTyped<TestData, TestData>(communicator) { req ->
            TestData.newBuilder().setIndex(req.index + 1).setMessage("echo").build()
        }

        val reqBytes = TestData.newBuilder().setIndex(7).setMessage("req").build().toByteArray()
        communicator.onReceivedData("TestData", reqBytes, incomingNonce = 42)

        var elapsed = 0
        while (transport.packets.isEmpty() && elapsed < 1000) {
            delay(5)
            elapsed += 5
        }

        assertTrue(transport.packets.isNotEmpty(), "expected a response packet")
        val packet = transport.packets.first()
        assertEquals(42, packet.responseNonce, "responseNonce should be 42")
        val payload = TestData.parseFrom(packet.data.toByteArray())
        assertEquals(8, payload.index, "response index should be 8")
        communicator.close()
    }
}
