feat: define and integrate STREAM_CODEC for enhanced packet serialization

- Added `STREAM_CODEC` to multiple packet classes (e.g., Clientbound, Serverbound) for consistent and efficient serialization.
- Replaced legacy codec implementations with `StreamCodec.composite` for standardization.
- Introduced `SealedCodecBuilder` for sealed class codec construction, improving flexibility and reducing boilerplate.
- Updated packet listener methods to handle newly implemented codecs (e.g., `handlePong`, `handleAttachedNote`).
- Refactored `ConnectionImpl` to skip processing responses handled by `RespondingPacketSendHandler`.
- Removed deprecated `@Serializable` annotation where unnecessary.

Author: twisti-dev

surf-cloud-api/surf-cloud-api-common/src/main/kotlin/dev/slne/surf/cloud/api/common/netty/network/codec/ByteBufCodecs.kt

@@ -516,6 +516,15 @@ object ByteBufCodecs {
         return CodecOperation { size -> collection(::ObjectArrayList, size) }
     }
 
+    fun <B : ByteBuf, V> makeImmutableList(): CodecOperation<B, MutableList<V>, List<V>> {
+        return CodecOperation { size ->
+            size.map(
+                { Collections.unmodifiableList(it) },
+                ::ObjectArrayList
+            )
+        }
+    }
+
     fun <B : ByteBuf, V> list(maxSize: Int): CodecOperation<B, V, MutableList<V>> {
         return CodecOperation { size -> collection(::ObjectArrayList, size, maxSize) }
     }

surf-cloud-api/surf-cloud-api-common/src/main/kotlin/dev/slne/surf/cloud/api/common/netty/network/codec/SealedCodecBuilder.kt

@@ -0,0 +1,35 @@
+package dev.slne.surf.cloud.api.common.netty.network.codec
+
+import dev.slne.surf.cloud.api.common.netty.protocol.buffer.checkDecodedNotNull
+import dev.slne.surf.cloud.api.common.netty.protocol.buffer.checkEncodedNotNull
+import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap
+
+class SealedCodecBuilder<B, T : Any>(private val idOf: (T) -> Int) {
+    private val entries = Int2ObjectOpenHashMap<StreamCodec<B, out T>>()
+
+    fun <S : T> variant(id: Int, codec: StreamCodec<B, S>) {
+        entries[id] = codec
+    }
+
+    fun build(idCodec: StreamCodec<B, Int>): StreamCodec<B, T> {
+        entries.trim()
+
+        return object : StreamCodec<B, T> {
+            override fun encode(buf: B, value: T) {
+                val id = idOf(value)
+                idCodec.encode(buf, id)
+                val codec = entries[id]
+                checkEncodedNotNull(codec) { "Unknown id: $id" }
+                @Suppress("UNCHECKED_CAST")
+                (codec as StreamCodec<B, T>).encode(buf, value)
+            }
+
+            override fun decode(buf: B): T {
+                val id = idCodec.decode(buf)
+                val codec = entries[id]
+                checkDecodedNotNull(codec) { "Unknown id: $id" }
+                return codec.decode(buf)
+            }
+        }
+    }
+}

surf-cloud-api/surf-cloud-api-common/src/main/kotlin/dev/slne/surf/cloud/api/common/netty/network/codec/SealedVariantCodecProvider.kt

@@ -0,0 +1,53 @@
+package dev.slne.surf.cloud.api.common.netty.network.codec
+
+import dev.slne.surf.cloud.api.common.util.annotation.InternalApi
+import io.netty.buffer.ByteBuf
+import kotlin.reflect.full.companionObjectInstance
+
+typealias SealedVariantCodecProviderBufComplex<T> = SealedVariantCodecProvider.Complex<ByteBuf, T>
+typealias SealedVariantCodecProviderBufSimple<T> = SealedVariantCodecProvider.Simple<ByteBuf, T>
+
+sealed interface SealedVariantCodecProvider<B, T : Any> {
+    val id: Int
+
+    interface Complex<B, T : Any> : SealedVariantCodecProvider<B, T> {
+        fun localCodec(parent: StreamCodec<B, T>): StreamCodec<B, out T>
+    }
+
+    interface Simple<B, T : Any> : SealedVariantCodecProvider<B, T> {
+        @Suppress("PropertyName")
+        val STREAM_CODEC: StreamCodec<B, out T>
+    }
+
+    companion object {
+        @InternalApi
+        data class Provider<B, T : Any>(
+            val id: Int,
+            val codecSupplier: (parent: StreamCodec<B, T>) -> StreamCodec<B, out T>
+        )
+
+        @InternalApi
+        fun <B, T : Any> findProvider(klass: Class<out T>): Provider<B, T>? {
+            val kClass = klass.kotlin
+            fun providerFromInstance(instance: Any): Provider<B, T>? {
+                if (instance is Simple<*, *>) {
+                    @Suppress("UNCHECKED_CAST")
+                    val simpleInstance = instance as Simple<B, T>
+                    return Provider(simpleInstance.id) { simpleInstance.STREAM_CODEC }
+                } else if (instance is Complex<*, *>) {
+                    @Suppress("UNCHECKED_CAST")
+                    val complexInstance = instance as Complex<B, T>
+                    return Provider(complexInstance.id, complexInstance::localCodec)
+                }
+
+                return null
+            }
+
+
+            kClass.objectInstance?.let { providerFromInstance(it) }?.let { return it }
+            kClass.companionObjectInstance?.let { providerFromInstance(it) }?.let { return it }
+
+            return null
+        }
+    }
+}

surf-cloud-api/surf-cloud-api-common/src/main/kotlin/dev/slne/surf/cloud/api/common/netty/network/codec/StreamCodec.kt

@@ -1,9 +1,15 @@
 package dev.slne.surf.cloud.api.common.netty.network.codec
 
+import dev.slne.surf.cloud.api.common.netty.protocol.buffer.checkEncoded
+import dev.slne.surf.cloud.api.common.netty.protocol.buffer.checkEncodedNotNull
 import io.netty.buffer.ByteBuf
+import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap
+import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap
+import java.lang.reflect.Modifier
 import java.util.function.BiFunction
 import java.util.function.Function
 import java.util.function.UnaryOperator
+import kotlin.reflect.KClass
 
 interface StreamCodec<B, V> : StreamDecoder<B, V>, StreamEncoder<B, V> {
     fun <O> apply(function: CodecOperation<B, V, O>) = function.apply(this)
@@ -65,6 +71,72 @@ interface StreamCodec<B, V> : StreamDecoder<B, V>, StreamEncoder<B, V> {
             }
         }
 
+        inline fun <B, T : Any> forSealedClass(
+            idCodec: StreamCodec<B, Int>,
+            noinline idOf: (T) -> Int,
+            crossinline build: SealedCodecBuilder<B, T>.(self: StreamCodec<B, T>) -> Unit
+        ): StreamCodec<B, T> = recursive { self ->
+            val b = SealedCodecBuilder<B, T>(idOf)
+            b.build(self)
+            b.build(idCodec)
+        }
+
+        fun <B, T : Any> forSealedClassAuto(
+            idCodec: StreamCodec<B, Int>,
+            root: KClass<T>
+        ): StreamCodec<B, T> = recursive<B, T> { parent ->
+            val idToCodec = Int2ObjectOpenHashMap<StreamCodec<B, out T>>()
+            val clazzToId = Object2IntOpenHashMap<Class<out T>>().apply {
+                defaultReturnValue(-1)
+            }
+
+            fun <T : Any> KClass<out T>.finalLeaves(): List<Class<out T>> {
+                fun walk(targetClass: Class<out T>, classes: MutableList<Class<out T>>) {
+                    val perms = targetClass.permittedSubclasses
+                    if (perms.isNullOrEmpty()) {
+                        classes += targetClass; return
+                    }
+                    perms.forEach { sub ->
+                        @Suppress("UNCHECKED_CAST")
+                        val subC = sub as Class<out T>
+                        if (Modifier.isFinal(subC.modifiers)) classes += subC
+                        else walk(subC, classes)
+                    }
+                }
+
+                val root = this.java
+                return buildList { walk(root, this) }
+            }
+
+            for (leaf in root.finalLeaves()) {
+                val provider = SealedVariantCodecProvider.findProvider<B, T>(leaf)
+                    ?: error("No SealedVariantCodecProvider on $leaf (object or companion)")
+                val codec = provider.codecSupplier(parent)
+                require(!idToCodec.containsKey(provider.id)) { "Duplicate id ${provider.id}" }
+                idToCodec[provider.id] = codec
+                clazzToId[leaf] = provider.id
+            }
+
+            ofMember(
+                { value, buf ->
+                    val id = clazzToId.getInt(value.javaClass)
+                    checkEncoded(id != -1) { "Unknown class ${value::class}" }
+                    val codec = idToCodec.get(id)
+                    checkEncodedNotNull(codec) { "Unknown id $id" }
+
+                    idCodec.encode(buf, id)
+                    @Suppress("UNCHECKED_CAST")
+                    (codec as StreamCodec<B, T>).encode(buf, value)
+                },
+                { buf ->
+                    val id = idCodec.decode(buf)
+                    val codec = idToCodec.get(id)
+                    checkEncodedNotNull(codec) { "Unknown id $id" }
+                    codec.decode(buf)
+                }
+            )
+        }
+
         @JvmStatic
         fun <B, C, T1> composite(
             codec: StreamCodec<in B, T1>,