Redesign notice handling and introduce NoticeProcessorPipeline part 3:

- Extract `SyncController`;
- Add super type `ProtocolStruct` to `ProtoBuf` and `JceStruct` to standardize `NoticeProcessorPipeline.process`
- and other implementations for pipeline

mirai-core-utils/src/commonMain/kotlin/StandardUtils.kt

@@ -241,6 +241,12 @@ public fun String.truncated(length: Int, truncated: String = "..."): String {
     } else this
 }
 
+/**
+ * Similar to [run] bot with [Unit] return type.
+ *
+ * You should not reference to [T] directly in the [block].
+ */
+// can convert to contextual receiver in the future, or there might be a stdlib function which we can delegate to.
 public inline fun <T> T.context(block: T.() -> Unit) {
     contract { callsInPlace(block, InvocationKind.EXACTLY_ONCE) }
     return block()

mirai-core/src/commonMain/kotlin/QQAndroidBot.kt

@@ -156,7 +156,8 @@ internal open class QQAndroidBot constructor(
         set(
             NoticeProcessorPipeline,
             NoticeProcessorPipelineImpl().apply {
-                registerProcessor(MsgInfoDecoder())
+                registerProcessor(MsgInfoDecoder(pipelineLogger))
+
                 registerProcessor(FriendNoticeProcessor(pipelineLogger))
                 registerProcessor(GroupListNoticeProcessor(pipelineLogger))
                 registerProcessor(GroupMessageProcessor())
@@ -217,6 +218,7 @@ internal open class QQAndroidBot constructor(
     open fun createNetworkLevelComponents(): ComponentStorage {
         return ConcurrentComponentStorage {
             set(BotClientHolder, BotClientHolderImpl(bot, networkLogger.subLogger("BotClientHolder")))
+            set(SyncController, SyncControllerImpl())
         }.withFallback(defaultBotLevelComponents)
     }
 

mirai-core/src/commonMain/kotlin/message/contextualBugReportException.kt

@@ -9,26 +9,26 @@
 
 package net.mamoe.mirai.internal.message
 
-import kotlin.contracts.ExperimentalContracts
 import kotlin.contracts.InvocationKind
 import kotlin.contracts.contract
 
+internal data class ContextualBugReportException(
+    override val message: String,
+    override val cause: Throwable?,
+) : IllegalStateException()
 
 internal fun contextualBugReportException(
     context: String,
     forDebug: String,
     e: Throwable? = null,
     additional: String = "",
-): IllegalStateException {
+): ContextualBugReportException {
-    return IllegalStateException(
+    return ContextualBugReportException(
         "在 $context 时遇到了意料之中的问题. 请完整复制此日志提交给 mirai: https://github.com/mamoe/mirai/issues/new   $additional 调试信息: $forDebug",
-        e
+        e,
     )
 }
 
-@OptIn(ExperimentalContracts::class)
-@Suppress("INVISIBLE_MEMBER", "INVISIBLE_REFERENCE", "RESULT_CLASS_IN_RETURN_TYPE")
-@kotlin.internal.InlineOnly
 internal inline fun <R> runWithBugReport(context: String, forDebug: () -> String, block: () -> R): R {
     contract {
         callsInPlace(block, InvocationKind.EXACTLY_ONCE)

mirai-core/src/commonMain/kotlin/network/Packet.kt

@@ -11,6 +11,7 @@ package net.mamoe.mirai.internal.network
 
 import net.mamoe.mirai.internal.AbstractBot
 import net.mamoe.mirai.internal.network.handler.logger
+import net.mamoe.mirai.internal.utils.io.ProtocolStruct
 import net.mamoe.mirai.utils.MiraiLogger
 
 /*
@@ -42,9 +43,11 @@ internal fun Collection<Packet>.toPacket(): Packet {
 }
 
 internal fun MultiPacket(delegate: Collection<Packet>): MultiPacket = MultiPacketImpl(delegate)
+internal fun MultiPacket(delegate: Packet): MultiPacket =
+    if (delegate is MultiPacket) delegate else MultiPacket(listOf(delegate))
 
 internal open class MultiPacketImpl(
-    val delegate: Collection<Packet>
+    val delegate: Collection<Packet>,
 ) : MultiPacket, Collection<Packet> by delegate {
 
     override fun toString(): String = delegate.joinToString(
@@ -56,6 +59,7 @@ internal open class MultiPacketImpl(
 
 
 internal class ParseErrorPacket(
+    val data: ProtocolStruct,
     val error: Throwable,
     val direction: Direction = Direction.TO_BOT_LOGGER,
 ) : Packet, Packet.NoLog {

mirai-core/src/commonMain/kotlin/network/QQAndroidClient.kt

@@ -11,7 +11,6 @@
 
 package net.mamoe.mirai.internal.network
 
-import kotlinx.atomicfu.AtomicBoolean
 import kotlinx.atomicfu.AtomicInt
 import kotlinx.atomicfu.atomic
 import kotlinx.io.core.String
@@ -21,7 +20,6 @@ import net.mamoe.mirai.internal.BotAccount
 import net.mamoe.mirai.internal.QQAndroidBot
 import net.mamoe.mirai.internal.network.components.AccountSecrets
 import net.mamoe.mirai.internal.network.components.SsoSession
-import net.mamoe.mirai.internal.network.protocol.SyncingCacheList
 import net.mamoe.mirai.internal.network.protocol.data.jce.FileStoragePushFSSvcList
 import net.mamoe.mirai.internal.network.protocol.packet.Tlv
 import net.mamoe.mirai.internal.utils.AtomicIntSeq
@@ -142,60 +140,6 @@ internal open class QQAndroidClient(
         }
     }
 
-    class MessageSvcSyncData {
-        val firstNotify: AtomicBoolean = atomic(true)
-        var latestMsgNewGroupTime: Long = currentTimeSeconds()
-        var latestMsgNewFriendTime: Long = currentTimeSeconds()
-
-        @Volatile
-        var syncCookie: ByteArray? = null
-        var pubAccountCookie = EMPTY_BYTE_ARRAY
-        var msgCtrlBuf: ByteArray = EMPTY_BYTE_ARRAY
-
-
-        internal data class PbGetMessageSyncId(
-            val uid: Long,
-            val sequence: Int,
-            val time: Int
-        )
-
-        val pbGetMessageCacheList = SyncingCacheList<PbGetMessageSyncId>()
-
-        internal data class SystemMsgNewSyncId(
-            val sequence: Long,
-            val time: Long
-        )
-
-        val systemMsgNewGroupCacheList = SyncingCacheList<SystemMsgNewSyncId>(10)
-        val systemMsgNewFriendCacheList = SyncingCacheList<SystemMsgNewSyncId>(10)
-
-
-        internal data class PbPushTransMsgSyncId(
-            val uid: Long,
-            val sequence: Int,
-            val time: Int
-        )
-
-        val pbPushTransMsgCacheList = SyncingCacheList<PbPushTransMsgSyncId>(10)
-
-        internal data class OnlinePushReqPushSyncId(
-            val uid: Long,
-            val sequence: Short,
-            val time: Long
-        )
-
-        val onlinePushReqPushCacheList = SyncingCacheList<OnlinePushReqPushSyncId>(50)
-
-        internal data class PendingGroupMessageReceiptSyncId(
-            val messageRandom: Int,
-        )
-
-        val pendingGroupMessageReceiptCacheList = SyncingCacheList<PendingGroupMessageReceiptSyncId>(50)
-    }
-
-
-    val syncingController = MessageSvcSyncData()
-
     var t150: Tlv? = null
     var rollbackSig: ByteArray? = null
     var ipFromT149: ByteArray? = null

mirai-core/src/commonMain/kotlin/network/components/NoticeProcessorPipeline.kt

@@ -12,10 +12,10 @@ package net.mamoe.mirai.internal.network.components
 import net.mamoe.mirai.internal.QQAndroidBot
 import net.mamoe.mirai.internal.message.contextualBugReportException
 import net.mamoe.mirai.internal.network.Packet
+import net.mamoe.mirai.internal.network.ParseErrorPacket
 import net.mamoe.mirai.internal.network.component.ComponentKey
 import net.mamoe.mirai.internal.network.component.ComponentStorage
 import net.mamoe.mirai.internal.network.notice.decoders.MsgType0x2DC
-import net.mamoe.mirai.internal.network.protocol.data.jce.MsgInfo
 import net.mamoe.mirai.internal.network.protocol.data.jce.MsgType0x210
 import net.mamoe.mirai.internal.network.protocol.data.jce.RequestPushStatus
 import net.mamoe.mirai.internal.network.protocol.data.proto.MsgComm
@@ -24,8 +24,11 @@ import net.mamoe.mirai.internal.network.protocol.data.proto.OnlinePushTrans.PbMs
 import net.mamoe.mirai.internal.network.protocol.data.proto.Structmsg
 import net.mamoe.mirai.internal.network.protocol.packet.chat.receive.MessageSvcPbGetMsg
 import net.mamoe.mirai.internal.network.protocol.packet.chat.receive.OnlinePushPbPushTransMsg
+import net.mamoe.mirai.internal.network.toPacket
+import net.mamoe.mirai.internal.utils.io.ProtocolStruct
 import net.mamoe.mirai.utils.TypeKey
 import net.mamoe.mirai.utils.TypeSafeMap
+import net.mamoe.mirai.utils.toDebugString
 import net.mamoe.mirai.utils.uncheckedCast
 import java.util.*
 import java.util.concurrent.ConcurrentLinkedQueue
@@ -34,21 +37,35 @@ import kotlin.concurrent.read
 import kotlin.concurrent.write
 import kotlin.reflect.KClass
 
+internal typealias ProcessResult = Collection<Packet>
+
 /**
  * Centralized processor pipeline for [MessageSvcPbGetMsg] and [OnlinePushPbPushTransMsg]
  */
 internal interface NoticeProcessorPipeline {
     fun registerProcessor(processor: NoticeProcessor)
 
-    suspend fun process(bot: QQAndroidBot, data: Any?, attributes: TypeSafeMap = TypeSafeMap()): Collection<Packet>
+    /**
+     * Process [data] into [Packet]s. Exceptions are wrapped into [ParseErrorPacket]
+     */
+    suspend fun process(bot: QQAndroidBot, data: ProtocolStruct, attributes: TypeSafeMap = TypeSafeMap()): ProcessResult
 
     companion object : ComponentKey<NoticeProcessorPipeline> {
         val ComponentStorage.noticeProcessorPipeline get() = get(NoticeProcessorPipeline)
+
+        @JvmStatic
+        suspend inline fun QQAndroidBot.processPacketThroughPipeline(
+            data: ProtocolStruct,
+            attributes: TypeSafeMap = TypeSafeMap(),
+        ): Packet {
+            return components.noticeProcessorPipeline.process(this, data, attributes).toPacket()
+        }
     }
 }
 
 internal interface PipelineContext {
     val bot: QQAndroidBot
+
     val attributes: TypeSafeMap
 
 
@@ -99,7 +116,7 @@ internal interface PipelineContext {
      *
      * @return result collected from processors. This would also have been collected to this context (where you call [fire]).
      */
-    suspend fun fire(data: Any?): Collection<Packet>
+    suspend fun fire(data: ProtocolStruct): ProcessResult
 
     companion object {
         val KEY_FROM_SYNC = TypeKey<Boolean>("fromSync")
@@ -109,7 +126,7 @@ internal interface PipelineContext {
 
 internal inline val PipelineContext.context get() = this
 
-internal class NoticeProcessorPipelineImpl : NoticeProcessorPipeline {
+internal open class NoticeProcessorPipelineImpl : NoticeProcessorPipeline {
     private val processors = ArrayList<NoticeProcessor>()
     private val processorsLock = ReentrantReadWriteLock()
 
@@ -147,22 +164,37 @@ internal class NoticeProcessorPipelineImpl : NoticeProcessorPipeline {
             this.collected.addAll(packets)
         }
 
-        override suspend fun fire(data: Any?): Collection<Packet> {
+        override suspend fun fire(data: ProtocolStruct): ProcessResult {
             return process(bot, data, attributes)
         }
     }
 
 
-    override suspend fun process(bot: QQAndroidBot, data: Any?, attributes: TypeSafeMap): Collection<Packet> {
+    override suspend fun process(bot: QQAndroidBot, data: ProtocolStruct, attributes: TypeSafeMap): ProcessResult {
         processorsLock.read {
             val context = ContextImpl(bot, attributes)
             for (processor in processors) {
+                kotlin.runCatching {
                     processor.process(context, data)
+                }.onFailure { e ->
+                    context.collect(
+                        ParseErrorPacket(
+                            data,
+                            IllegalStateException(
+                                "Exception in $processor while processing packet ${packetToString(data)}.",
+                                e,
+                            ),
+                        ),
+                    )
+                }
             }
             return context.collected
         }
     }
 
+    protected open fun packetToString(data: Any?): String =
+        data.toDebugString("mirai.network.debug.notice.pipeline.log.full")
+
 }
 
 ///////////////////////////////////////////////////////////////////////////
@@ -203,7 +235,6 @@ internal abstract class MsgCommonMsgProcessor : SimpleNoticeProcessor<MsgComm.Ms
 internal abstract class MixedNoticeProcessor : AnyNoticeProcessor() {
     final override suspend fun PipelineContext.processImpl(data: Any) {
         when (data) {
-            is MsgInfo -> processImpl(data)
             is PbMsgInfo -> processImpl(data)
             is MsgOnlinePush.PbPushMsg -> processImpl(data)
             is MsgComm.Msg -> processImpl(data)
@@ -214,7 +245,6 @@ internal abstract class MixedNoticeProcessor : AnyNoticeProcessor() {
         }
     }
 
-    protected open suspend fun PipelineContext.processImpl(data: MsgInfo) {}
     protected open suspend fun PipelineContext.processImpl(data: MsgType0x210) {} // 528
     protected open suspend fun PipelineContext.processImpl(data: MsgType0x2DC) {} // 732
     protected open suspend fun PipelineContext.processImpl(data: PbMsgInfo) {}

mirai-core/src/commonMain/kotlin/network/components/PacketLoggingStrategy.kt

@@ -54,7 +54,7 @@ internal class PacketLoggingStrategyImpl(
                 packet ?: return
                 if (!bot.logger.isEnabled && !logger.isEnabled) return
                 if (packet is ParseErrorPacket) {
-                    packet.direction.getLogger(bot).error("Exception in parsing packet.", packet.error)
+                    packet.direction.getLogger(bot).error("Exception on parsing packet.", packet.error)
                 }
 
                 if (packet is MultiPacket) {
@@ -64,7 +64,7 @@ internal class PacketLoggingStrategyImpl(
                 }
 
                 logReceivedImpl(packet, incomingPacket, logger)
-            }
+            },
         )
     }
 

mirai-core/src/commonMain/kotlin/network/components/SyncController.kt

@@ -0,0 +1,143 @@
+/*
+ * Copyright 2019-2021 Mamoe Technologies and contributors.
+ *
+ *  此源代码的使用受 GNU AFFERO GENERAL PUBLIC LICENSE version 3 许可证的约束, 可以在以下链接找到该许可证.
+ *  Use of this source code is governed by the GNU AGPLv3 license that can be found through the following link.
+ *
+ *  https://github.com/mamoe/mirai/blob/master/LICENSE
+ */
+
+package net.mamoe.mirai.internal.network.components
+
+import kotlinx.atomicfu.AtomicBoolean
+import kotlinx.atomicfu.atomic
+import net.mamoe.mirai.internal.AbstractBot
+import net.mamoe.mirai.internal.network.QQAndroidClient
+import net.mamoe.mirai.internal.network.component.ComponentKey
+import net.mamoe.mirai.internal.network.protocol.SyncingCacheList
+import net.mamoe.mirai.internal.network.protocol.data.jce.MsgInfo
+import net.mamoe.mirai.internal.network.protocol.data.proto.MsgComm
+import net.mamoe.mirai.internal.network.protocol.data.proto.OnlinePushTrans
+import net.mamoe.mirai.utils.EMPTY_BYTE_ARRAY
+import net.mamoe.mirai.utils.currentTimeSeconds
+
+internal interface SyncController {
+    val firstNotify: AtomicBoolean
+    var latestMsgNewGroupTime: Long
+    var latestMsgNewFriendTime: Long
+
+    var syncCookie: ByteArray?
+    var pubAccountCookie: ByteArray
+    var msgCtrlBuf: ByteArray
+
+    fun syncOnlinePush(uid: Long, sequence: Short, time: Long): Boolean
+    fun syncNewFriend(sequence: Long, time: Long): Boolean
+    fun syncNewGroup(sequence: Long, time: Long): Boolean
+    fun syncGetMessage(uid: Long, sequence: Int, time: Int): Boolean
+    fun syncPushTrans(uid: Long, sequence: Int, time: Int): Boolean
+
+    fun syncGroupMessageReceipt(messageRandom: Int): Boolean
+    fun containsGroupMessageReceipt(messageRandom: Int): Boolean
+
+    companion object : ComponentKey<SyncController> {
+        val AbstractBot.syncController get() = this.components[SyncController]
+        val QQAndroidClient.syncController get() = bot.syncController
+        var QQAndroidClient.syncCookie
+            get() = bot.syncController.syncCookie
+            set(value) {
+                bot.syncController.syncCookie = value
+            }
+    }
+}
+
+internal fun SyncController.syncPushTrans(content: OnlinePushTrans.PbMsgInfo): Boolean =
+    syncPushTrans(content.msgUid, content.msgSeq, content.msgTime)
+
+internal fun SyncController.syncGetMessage(
+    msgHead: MsgComm.MsgHead,
+) = msgHead.run {
+    syncGetMessage(msgUid, msgSeq, msgTime)
+}
+
+internal fun SyncController.syncOnlinePush(
+    msgInfo: MsgInfo,
+) = syncOnlinePush(
+    uid = msgInfo.lMsgUid ?: 0,
+    sequence = msgInfo.shMsgSeq,
+    time = msgInfo.uMsgTime,
+)
+
+internal class SyncControllerImpl : SyncController {
+    override val firstNotify: AtomicBoolean = atomic(true)
+
+    @Volatile
+    override var latestMsgNewGroupTime: Long = currentTimeSeconds()
+
+    @Volatile
+    override var latestMsgNewFriendTime: Long = currentTimeSeconds()
+
+    @Volatile
+    override var syncCookie: ByteArray? = null
+
+    @Volatile
+    override var pubAccountCookie = EMPTY_BYTE_ARRAY
+
+    @Volatile
+    override var msgCtrlBuf: ByteArray = EMPTY_BYTE_ARRAY
+
+    private val pbGetMessageCacheList = SyncingCacheList<PbGetMessageSyncId>()
+    private val systemMsgNewGroupCacheList = SyncingCacheList<SystemMsgNewSyncId>(10)
+    private val systemMsgNewFriendCacheList = SyncingCacheList<SystemMsgNewSyncId>(10)
+    private val pbPushTransMsgCacheList = SyncingCacheList<PbPushTransMsgSyncId>(10)
+    private val onlinePushReqPushCacheList = SyncingCacheList<OnlinePushReqPushSyncId>(50)
+    private val pendingGroupMessageReceiptCacheList = SyncingCacheList<PendingGroupMessageReceiptSyncId>(50)
+
+    override fun syncOnlinePush(uid: Long, sequence: Short, time: Long): Boolean =
+        onlinePushReqPushCacheList.addCache(OnlinePushReqPushSyncId(uid, sequence, time))
+
+    override fun syncNewFriend(sequence: Long, time: Long): Boolean =
+        systemMsgNewFriendCacheList.addCache(SystemMsgNewSyncId(sequence, time))
+
+    override fun syncNewGroup(sequence: Long, time: Long): Boolean =
+        systemMsgNewGroupCacheList.addCache(SystemMsgNewSyncId(sequence, time))
+
+    override fun syncGetMessage(uid: Long, sequence: Int, time: Int): Boolean =
+        pbGetMessageCacheList.addCache(PbGetMessageSyncId(uid, sequence, time))
+
+    override fun syncPushTrans(uid: Long, sequence: Int, time: Int): Boolean =
+        pbPushTransMsgCacheList.addCache(PbPushTransMsgSyncId(uid, sequence, time))
+
+    override fun syncGroupMessageReceipt(messageRandom: Int): Boolean =
+        pendingGroupMessageReceiptCacheList.addCache(PendingGroupMessageReceiptSyncId(messageRandom))
+
+    override fun containsGroupMessageReceipt(messageRandom: Int): Boolean =
+        pendingGroupMessageReceiptCacheList.contains { it.messageRandom == messageRandom }
+
+    data class PbGetMessageSyncId(
+        val uid: Long,
+        val sequence: Int,
+        val time: Int,
+    )
+
+    data class SystemMsgNewSyncId(
+        val sequence: Long,
+        val time: Long,
+    )
+
+    data class PbPushTransMsgSyncId(
+        val uid: Long,
+        val sequence: Int,
+        val time: Int,
+    )
+
+    data class OnlinePushReqPushSyncId(
+        val uid: Long,
+        val sequence: Short,
+        val time: Long,
+    )
+
+    data class PendingGroupMessageReceiptSyncId(
+        val messageRandom: Int,
+    )
+
+}

mirai-core/src/commonMain/kotlin/network/notice/GroupListNoticeProcessor.kt

@@ -197,8 +197,8 @@ internal class GroupListNoticeProcessor(
     ///////////////////////////////////////////////////////////////////////////
 
     override suspend fun PipelineContext.processImpl(data: Structmsg.StructMsg) = data.msg.context {
-        markAsConsumed()
         if (this == null) return
+        markAsConsumed()
         when (subType) {
             // 处理被邀请入群 或 处理成员入群申请
             1 -> when (groupMsgType) {
@@ -259,16 +259,12 @@ internal class GroupListNoticeProcessor(
                             "解析 NewContact.SystemMsgNewGroup, subType=5, groupMsgType=$groupMsgType",
                             this._miraiContentToString(),
                             null,
-                            "并描述此时机器人是否被踢出群等"
+                            "并描述此时机器人是否被踢出群等",
                         )
                     }
                 }
             }
-            else -> throw contextualBugReportException(
+            else -> markNotConsumed()
-                "解析 NewContact.SystemMsgNewGroup, subType=$subType, groupMsgType=$groupMsgType",
-                forDebug = this._miraiContentToString(),
-                additional = "并尽量描述此时机器人是否正被邀请加入群, 或者是有有新群员加入此群"
-            )
         }
     }
 

mirai-core/src/commonMain/kotlin/network/notice/GroupMessageProcessor.kt

@@ -25,6 +25,7 @@ import net.mamoe.mirai.internal.message.toMessageChainOnline
 import net.mamoe.mirai.internal.network.Packet
 import net.mamoe.mirai.internal.network.components.PipelineContext
 import net.mamoe.mirai.internal.network.components.SimpleNoticeProcessor
+import net.mamoe.mirai.internal.network.components.SyncController.Companion.syncController
 import net.mamoe.mirai.internal.network.handler.logger
 import net.mamoe.mirai.internal.network.notice.GroupMessageProcessor.MemberNick.Companion.generateMemberNickFromMember
 import net.mamoe.mirai.internal.network.protocol.data.proto.ImMsgBody
@@ -71,8 +72,9 @@ internal class GroupMessageProcessor : SimpleNoticeProcessor<MsgOnlinePush.PbPus
         if (isFromSelfAccount) {
             val messageRandom = data.msg.msgBody.richText.attr?.random ?: return
 
-            if (bot.client.syncingController.pendingGroupMessageReceiptCacheList.contains { it.messageRandom == messageRandom }
+            if (bot.syncController.containsGroupMessageReceipt(messageRandom)
-                || msgHead.fromAppid == 3116 || msgHead.fromAppid == 2021) {
+                || msgHead.fromAppid == 3116 || msgHead.fromAppid == 2021
+            ) {
                 // 3116=group music share
                 // 2021=group file
                 // message sent by bot

mirai-core/src/commonMain/kotlin/network/notice/OtherClientNoticeProcessor.kt

@@ -94,11 +94,7 @@ internal class OtherClientNoticeProcessor : MixedNoticeProcessor() {
                     collected += OtherClientOfflineEvent(client)
                 }
 
-                else -> throw contextualBugReportException(
+                else -> markNotConsumed()
-                    "decode SvcRequestPushStatus (PC Client status change)",
-                    data._miraiContentToString(),
-                    additional = "unknown status=${data.status}",
-                )
             }
         }
     }

mirai-core/src/commonMain/kotlin/network/notice/PrivateMessageNoticeProcessor.kt

@@ -56,9 +56,7 @@ internal class PrivateMessageNoticeProcessor : SimpleNoticeProcessor<MsgComm.Msg
                 val group = bot.getGroupByUin(tmpHead.groupUin) ?: return
                 handlePrivateMessage(data, group[senderUin] ?: return)
             }
-            else -> {
+            else -> markNotConsumed()
-                markNotConsumed()
-            }
         }
 
     }

mirai-core/src/commonMain/kotlin/network/notice/UnconsumedNoticesAlerter.kt

@@ -13,7 +13,6 @@ import net.mamoe.mirai.internal.message.contextualBugReportException
 import net.mamoe.mirai.internal.network.components.MixedNoticeProcessor
 import net.mamoe.mirai.internal.network.components.PipelineContext
 import net.mamoe.mirai.internal.network.notice.decoders.MsgType0x2DC
-import net.mamoe.mirai.internal.network.protocol.data.jce.MsgInfo
 import net.mamoe.mirai.internal.network.protocol.data.jce.MsgType0x210
 import net.mamoe.mirai.internal.network.protocol.data.jce.RequestPushStatus
 import net.mamoe.mirai.internal.network.protocol.data.proto.MsgComm
@@ -23,17 +22,13 @@ import net.mamoe.mirai.internal.network.protocol.data.proto.Structmsg
 import net.mamoe.mirai.internal.network.protocol.packet.chat.NewContact
 import net.mamoe.mirai.internal.utils._miraiContentToString
 import net.mamoe.mirai.utils.MiraiLogger
+import net.mamoe.mirai.utils.context
 import net.mamoe.mirai.utils.debug
 import net.mamoe.mirai.utils.toUHexString
 
 internal class UnconsumedNoticesAlerter(
     private val logger: MiraiLogger,
 ) : MixedNoticeProcessor() {
-    override suspend fun PipelineContext.processImpl(data: MsgInfo) {
-        if (isConsumed) return
-        logger.debug { "Unknown kind ${data.shMsgType.toInt()}, data=${data.vMsg.toUHexString()}" }
-    }
-
     override suspend fun PipelineContext.processImpl(data: MsgType0x210) {
         if (isConsumed) return
         when (data.uSubMsgType) {
@@ -42,7 +37,6 @@ internal class UnconsumedNoticesAlerter(
             0xD4L, // bot 在其他客户端被踢或主动退出而同步情况
             -> {
                 // Network(1994701021) 16:03:54 : unknown group 528 type 0x0000000000000026, data: 08 01 12 40 0A 06 08 F4 EF BB 8F 04 10 E7 C1 AD B8 02 18 01 22 2C 10 01 1A 1A 18 B4 DC F8 9B 0C 20 E7 C1 AD B8 02 28 06 30 02 A2 01 04 08 93 D6 03 A8 01 08 20 00 28 00 32 08 18 01 20 FE AF AF F5 05 28 00
-
             }
 
             0xE2L -> {
@@ -51,10 +45,11 @@ internal class UnconsumedNoticesAlerter(
                 // 0A 35 08 00 10 A2 FF 8C F0 03 1A 1B E5 90 8C E6 84 8F E4 BD A0 E7 9A 84 E5 8A A0 E5 A5 BD E5 8F 8B E8 AF B7 E6 B1 82 22 0C E6 BD 9C E6 B1 9F E7 BE A4 E5 8F 8B 28 01
                 // vProtobuf.loadAs(Msgtype0x210.serializer())
             }
-
+            else -> {
-        }
                 logger.debug { "Unknown group 528 type 0x${data.uSubMsgType.toUHexString("")}, data: " + data.vProtobuf.toUHexString() }
             }
+        }
+    }
 
     override suspend fun PipelineContext.processImpl(data: MsgType0x2DC) {
         if (isConsumed) return
@@ -125,11 +120,21 @@ internal class UnconsumedNoticesAlerter(
 
     override suspend fun PipelineContext.processImpl(data: Structmsg.StructMsg) {
         if (isConsumed) return
-        TODO("Not yet implemented")
+        data.msg?.context {
+            throw contextualBugReportException(
+                "解析 NewContact.SystemMsgNewGroup, subType=$subType, groupMsgType=$groupMsgType",
+                forDebug = this._miraiContentToString(),
+                additional = "并尽量描述此时机器人是否正被邀请加入群, 或者是有有新群员加入此群",
+            )
+        }
     }
 
     override suspend fun PipelineContext.processImpl(data: RequestPushStatus) {
         if (isConsumed) return
-        TODO("Not yet implemented")
+        throw contextualBugReportException(
+            "decode SvcRequestPushStatus (PC Client status change)",
+            data._miraiContentToString(),
+            additional = "unknown status=${data.status}",
+        )
     }
 }

mirai-core/src/commonMain/kotlin/network/notice/decoders/MsgInfoDecoder.kt

@@ -16,17 +16,34 @@ import net.mamoe.mirai.internal.contact.GroupImpl
 import net.mamoe.mirai.internal.contact.checkIsGroupImpl
 import net.mamoe.mirai.internal.network.components.PipelineContext
 import net.mamoe.mirai.internal.network.components.SimpleNoticeProcessor
+import net.mamoe.mirai.internal.network.components.SyncController.Companion.syncController
+import net.mamoe.mirai.internal.network.components.syncOnlinePush
 import net.mamoe.mirai.internal.network.protocol.data.jce.MsgInfo
 import net.mamoe.mirai.internal.network.protocol.data.jce.MsgType0x210
+import net.mamoe.mirai.internal.network.protocol.data.jce.OnlinePushPack.SvcReqPushMsg
+import net.mamoe.mirai.internal.utils.io.ProtocolStruct
 import net.mamoe.mirai.internal.utils.io.serialization.loadAs
+import net.mamoe.mirai.utils.MiraiLogger
+import net.mamoe.mirai.utils.debug
 import net.mamoe.mirai.utils.read
+import net.mamoe.mirai.utils.toUHexString
 
 /**
- * Decodes [MsgInfo] and re-fire [MsgType0x210] or [MsgType0x2DC]
+ * Decodes [SvcReqPushMsg] to [MsgInfo] then re-fire [MsgType0x210] or [MsgType0x2DC]
  */
-internal class MsgInfoDecoder : SimpleNoticeProcessor<MsgInfo>(type()) {
+internal class MsgInfoDecoder(
-    override suspend fun PipelineContext.processImpl(data: MsgInfo) {
+    private val logger: MiraiLogger,
-        markAsConsumed()
+) : SimpleNoticeProcessor<SvcReqPushMsg>(type()) {
+    override suspend fun PipelineContext.processImpl(data: SvcReqPushMsg) {
+        // SvcReqPushMsg is fully handled here, no need to set consumed.
+
+        for (msgInfo in data.vMsgInfos) {
+            decodeMsgInfo(msgInfo)
+        }
+    }
+
+    private suspend fun PipelineContext.decodeMsgInfo(data: MsgInfo) {
+        if (!bot.syncController.syncOnlinePush(data)) return
         when (data.shMsgType.toUShort().toInt()) {
             // 528
             0x210 -> fire(data.vMsg.loadAs(MsgType0x210.serializer()))
@@ -43,7 +60,9 @@ internal class MsgInfoDecoder : SimpleNoticeProcessor<MsgInfo>(type()) {
                     fire(MsgType0x2DC(kind, group, this.readBytes()))
                 }
             }
-            else -> markNotConsumed()
+            else -> {
+                logger.debug { "Unknown MsgInfo kind ${data.shMsgType.toInt()}, data=${data.vMsg.toUHexString()}" }
+            }
         }
     }
 }
@@ -52,4 +71,4 @@ internal class MsgType0x2DC(
     val kind: Int, // inner kind, read from vMsg
     val group: GroupImpl,
     val buf: ByteArray,
-)
+) : ProtocolStruct

mirai-core/src/commonMain/kotlin/network/protocol/packet/PacketFactory.kt

@@ -13,7 +13,6 @@ import kotlinx.io.core.ByteReadPacket
 import net.mamoe.mirai.event.Event
 import net.mamoe.mirai.internal.QQAndroidBot
 import net.mamoe.mirai.internal.network.Packet
-import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.noticeProcessorPipeline
 import net.mamoe.mirai.internal.network.components.PacketCodec
 import net.mamoe.mirai.internal.network.protocol.packet.chat.*
 import net.mamoe.mirai.internal.network.protocol.packet.chat.image.ImgStore
@@ -28,9 +27,7 @@ import net.mamoe.mirai.internal.network.protocol.packet.login.Heartbeat
 import net.mamoe.mirai.internal.network.protocol.packet.login.StatSvc
 import net.mamoe.mirai.internal.network.protocol.packet.login.WtLogin
 import net.mamoe.mirai.internal.network.protocol.packet.summarycard.SummaryCard
-import net.mamoe.mirai.internal.network.toPacket
 import net.mamoe.mirai.utils.MiraiLoggerWithSwitch
-import net.mamoe.mirai.utils.TypeSafeMap
 
 internal sealed class PacketFactory<TPacket : Packet?> {
     /**
@@ -39,17 +36,6 @@ internal sealed class PacketFactory<TPacket : Packet?> {
     abstract val receivingCommandName: String
 
     open val canBeCached: Boolean get() = true
-
-    protected companion object {
-
-        @JvmStatic
-        suspend fun QQAndroidBot.processPacketThroughPipeline(
-            data: Any?,
-            attributes: TypeSafeMap = TypeSafeMap(),
-        ): Packet {
-            return components.noticeProcessorPipeline.process(this, data, attributes).toPacket()
-        }
-    }
 }
 
 /**

mirai-core/src/commonMain/kotlin/network/protocol/packet/chat/NewContact.kt

@@ -15,19 +15,23 @@ import kotlinx.io.core.ByteReadPacket
 import kotlinx.io.core.readBytes
 import net.mamoe.mirai.event.events.NewFriendRequestEvent
 import net.mamoe.mirai.internal.QQAndroidBot
-import net.mamoe.mirai.internal.network.*
+import net.mamoe.mirai.internal.network.Packet
-import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.noticeProcessorPipeline
+import net.mamoe.mirai.internal.network.QQAndroidClient
+import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.processPacketThroughPipeline
+import net.mamoe.mirai.internal.network.components.SyncController.Companion.syncController
 import net.mamoe.mirai.internal.network.protocol.data.proto.Structmsg
 import net.mamoe.mirai.internal.network.protocol.packet.OutgoingPacketFactory
 import net.mamoe.mirai.internal.network.protocol.packet.buildOutgoingUniPacket
+import net.mamoe.mirai.internal.network.toPacket
 import net.mamoe.mirai.internal.utils.io.serialization.loadAs
+import net.mamoe.mirai.internal.utils.io.serialization.readProtoBuf
 import net.mamoe.mirai.internal.utils.io.serialization.writeProtoBuf
 import kotlin.math.max
 
 internal class NewContact {
 
     internal object SystemMsgNewFriend :
-        OutgoingPacketFactory<Packet?>("ProfileService.Pb.ReqSystemMsgNew.Friend") {
+        OutgoingPacketFactory<Packet>("ProfileService.Pb.ReqSystemMsgNew.Friend") {
 
         operator fun invoke(client: QQAndroidClient) = buildOutgoingUniPacket(client) {
             writeProtoBuf(
@@ -39,30 +43,24 @@ internal class NewContact {
                         frdMsgGetBusiCard = 1,
                         frdMsgNeedWaitingMsg = 1,
                         frdMsgUint32NeedAllUnreadMsg = 1,
-                        grpMsgMaskInviteAutoJoin = 1
+                        grpMsgMaskInviteAutoJoin = 1,
                     ),
                     friendMsgTypeFlag = 1,
                     isGetFrdRibbon = false,
                     isGetGrpRibbon = false,
                     msgNum = 20,
-                    version = 1000
+                    version = 1000,
-                )
+                ),
             )
         }
 
 
-        override suspend fun ByteReadPacket.decode(bot: QQAndroidBot): Packet? {
+        override suspend fun ByteReadPacket.decode(bot: QQAndroidBot): Packet {
-            readBytes().loadAs(Structmsg.RspSystemMsgNew.serializer()).run {
+            readProtoBuf(Structmsg.RspSystemMsgNew.serializer()).run {
                 return friendmsgs.filter {
-                    it.msgTime >= bot.client.syncingController.latestMsgNewFriendTime
+                    it.msgTime >= bot.syncController.latestMsgNewFriendTime
                 }.mapNotNull { struct ->
-                    if (!bot.client.syncingController.systemMsgNewFriendCacheList.addCache(
+                    if (!bot.syncController.syncNewFriend(struct.msgSeq, struct.msgTime)) { // duplicate
-                            QQAndroidClient.MessageSvcSyncData.SystemMsgNewSyncId(
-                                struct.msgSeq,
-                                struct.msgTime
-                            )
-                        )
-                    ) { // duplicate
                         return@mapNotNull null
                     }
                     struct.msg?.run {
@@ -72,17 +70,11 @@ internal class NewContact {
                             msgAdditional,
                             struct.reqUin,
                             groupCode,
-                            reqUinNick
+                            reqUinNick,
                         )
                     }
-                }.let { packets ->
+                }.toPacket().also {
-                    when {
+                    bot.syncController.run {
-                        packets.isEmpty() -> null
-                        packets.size == 1 -> packets[0]
-                        else -> MultiPacket(packets)
-                    }
-                }.also {
-                    bot.client.syncingController.run {
                         latestMsgNewFriendTime = max(latestMsgNewFriendTime, friendmsgs.maxOfOrNull { it.msgTime } ?: 0)
                     }
                 }
@@ -96,7 +88,7 @@ internal class NewContact {
                 eventId: Long,
                 fromId: Long,
                 accept: Boolean,
-                blackList: Boolean = false
+                blackList: Boolean = false,
             ) =
                 buildOutgoingUniPacket(client) {
                     writeProtoBuf(
@@ -107,14 +99,14 @@ internal class NewContact {
                                 addFrdSNInfo = Structmsg.AddFrdSNInfo(),
                                 msg = "",
                                 remark = "",
-                                blacklist = !accept && blackList
+                                blacklist = !accept && blackList,
                             ),
                             msgSeq = eventId,
                             reqUin = fromId,
                             srcId = 6,
                             subSrcId = 7,
-                            subType = 1
+                            subType = 1,
-                        )
+                        ),
                     )
                 }
 
@@ -148,39 +140,29 @@ internal class NewContact {
                         grpMsgNeedAutoAdminWording = 1,
                         grpMsgNotAllowJoinGrpInviteNotFrd = 1,
                         grpMsgSupportInviteAutoJoin = 1,
-                        grpMsgWordingDown = 1
+                        grpMsgWordingDown = 1,
                     ),
                     friendMsgTypeFlag = 1,
                     isGetFrdRibbon = false,
                     isGetGrpRibbon = false,
                     msgNum = 5,
-                    version = 1000
+                    version = 1000,
-                )
+                ),
             )
         }
 
 
-        override suspend fun ByteReadPacket.decode(bot: QQAndroidBot): Packet? {
+        override suspend fun ByteReadPacket.decode(bot: QQAndroidBot): Packet {
             return readBytes().loadAs(Structmsg.RspSystemMsgNew.serializer()).run {
                 groupmsgs.filter {
-                    it.msgTime >= bot.client.syncingController.latestMsgNewGroupTime
+                    it.msgTime >= bot.syncController.latestMsgNewGroupTime
                 }.mapNotNull { struct ->
-                    if (!bot.client.syncingController.systemMsgNewGroupCacheList.addCache(
+                    if (!bot.syncController.syncNewGroup(struct.msgSeq, struct.msgTime)) { // duplicate
-                            QQAndroidClient.MessageSvcSyncData.SystemMsgNewSyncId(
-                                struct.msgSeq,
-                                struct.msgTime
-                            )
-                        )
-                    ) { // duplicate
                         return@mapNotNull null
                     }
-                    try {
+                    bot.processPacketThroughPipeline(struct)
-                        bot.components.noticeProcessorPipeline.process(bot, struct).toPacket()
-                    } catch (e: Throwable) {
-                        ParseErrorPacket(e)
-                    }
                 }.toPacket().also {
-                    bot.client.syncingController.run {
+                    bot.syncController.run {
                         latestMsgNewGroupTime = max(latestMsgNewGroupTime, groupmsgs.maxOfOrNull { it.msgTime } ?: 0)
                     }
                 }
@@ -197,7 +179,7 @@ internal class NewContact {
                 isInvited: Boolean,
                 accept: Boolean?,
                 blackList: Boolean = false,
-                message: String = ""
+                message: String = "",
             ) =
                 buildOutgoingUniPacket(client) {
                     writeProtoBuf(
@@ -212,7 +194,7 @@ internal class NewContact {
                                 groupCode = groupId,
                                 msg = message,
                                 remark = "",
-                                blacklist = blackList
+                                blacklist = blackList,
                             ),
                             groupMsgType = if (isInvited) 2 else 1,
                             language = 1000,
@@ -220,8 +202,8 @@ internal class NewContact {
                             reqUin = fromId,
                             srcId = 3,
                             subSrcId = if (isInvited) 10016 else 31,
-                            subType = 1
+                            subType = 1,
-                        )
+                        ),
                     )
                 }
 

mirai-core/src/commonMain/kotlin/network/protocol/packet/chat/receive/MessageSvc.PbGetMsg.kt

@@ -22,13 +22,17 @@ import net.mamoe.mirai.internal.QQAndroidBot
 import net.mamoe.mirai.internal.network.MultiPacket
 import net.mamoe.mirai.internal.network.Packet
 import net.mamoe.mirai.internal.network.QQAndroidClient
+import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.processPacketThroughPipeline
 import net.mamoe.mirai.internal.network.components.PipelineContext.Companion.KEY_FROM_SYNC
+import net.mamoe.mirai.internal.network.components.SyncController.Companion.syncController
+import net.mamoe.mirai.internal.network.components.syncGetMessage
 import net.mamoe.mirai.internal.network.protocol.data.proto.MsgComm
 import net.mamoe.mirai.internal.network.protocol.data.proto.MsgSvc
 import net.mamoe.mirai.internal.network.protocol.packet.OutgoingPacketFactory
 import net.mamoe.mirai.internal.network.protocol.packet.buildOutgoingUniPacket
 import net.mamoe.mirai.internal.utils.io.serialization.readProtoBuf
 import net.mamoe.mirai.internal.utils.io.serialization.writeProtoBuf
+import net.mamoe.mirai.utils.toLongUnsigned
 import kotlin.random.Random
 
 
@@ -57,7 +61,7 @@ internal object MessageSvcPbGetMsg : OutgoingPacketFactory<MessageSvcPbGetMsg.Re
                 whisperSessionId = 0,
                 syncFlag = syncFlag,
                 //  serverBuf = from.serverBuf ?: EMPTY_BYTE_ARRAY,
-                syncCookie = syncCookie ?: client.syncingController.syncCookie
+                syncCookie = syncCookie ?: client.bot.syncController.syncCookie
                 ?: byteArrayOf(), //.also { client.c2cMessageSync.syncCookie = it },
                 // syncFlag = client.c2cMessageSync.syncFlag,
                 //msgCtrlBuf = client.c2cMessageSync.msgCtrlBuf,
@@ -113,14 +117,14 @@ internal object MessageSvcPbGetMsg : OutgoingPacketFactory<MessageSvcPbGetMsg.Re
         }
         when (resp.msgRspType) {
             0 -> {
-                bot.client.syncingController.syncCookie = resp.syncCookie
+                bot.syncController.syncCookie = resp.syncCookie
-                bot.client.syncingController.pubAccountCookie = resp.pubAccountCookie
+                bot.syncController.pubAccountCookie = resp.pubAccountCookie
             }
             1 -> {
-                bot.client.syncingController.syncCookie = resp.syncCookie
+                bot.syncController.syncCookie = resp.syncCookie
             }
             2 -> {
-                bot.client.syncingController.pubAccountCookie = resp.pubAccountCookie
+                bot.syncController.pubAccountCookie = resp.pubAccountCookie
 
             }
         }
@@ -129,33 +133,20 @@ internal object MessageSvcPbGetMsg : OutgoingPacketFactory<MessageSvcPbGetMsg.Re
         // bot.logger.debug(resp.msgRspType._miraiContentToString())
         // bot.logger.debug(resp.syncCookie._miraiContentToString())
 
-        bot.client.syncingController.msgCtrlBuf = resp.msgCtrlBuf
+        bot.syncController.msgCtrlBuf = resp.msgCtrlBuf
 
         if (resp.uinPairMsgs.isEmpty()) return EmptyResponse(bot)
 
-        for (uinPairMsg in resp.uinPairMsgs) {
-            if (uinPairMsg.msg.isEmpty()) continue
-            for (msg in uinPairMsg.msg) {
-                if (msg.msgHead.msgTime <= uinPairMsg.lastReadTime.toLong() and 4294967295L) continue
-            }
-        }
-
         val messages = resp.uinPairMsgs.asSequence()
             .filterNot { it.msg.isEmpty() }
             .flatMap { pair ->
                 pair.msg.asSequence()
-                    .filter { msg: MsgComm.Msg -> msg.msgHead.msgTime > pair.lastReadTime.toLong() and 4294967295L }
+                    .filter { msg: MsgComm.Msg -> msg.msgHead.msgTime > pair.lastReadTime.toLongUnsigned() }
             }
             .toList()
             .also { MessageSvcPbDeleteMsg.delete(bot, it) } // 删除消息
             .filter { msg ->
-                bot.client.syncingController.pbGetMessageCacheList.addCache(
+                bot.syncController.syncGetMessage(msg.msgHead)
-                    QQAndroidClient.MessageSvcSyncData.PbGetMessageSyncId(
-                        uid = msg.msgHead.msgUid,
-                        sequence = msg.msgHead.msgSeq,
-                        time = msg.msgHead.msgTime,
-                    ),
-                )
             }
             .map { msg ->
                 bot.processPacketThroughPipeline(msg, KEY_FROM_SYNC to false)
@@ -179,7 +170,7 @@ internal object MessageSvcPbGetMsg : OutgoingPacketFactory<MessageSvcPbGetMsg.Re
                     MessageSvcPbGetMsg(
                         client,
                         MsgSvc.SyncFlag.CONTINUE,
-                        bot.client.syncingController.syncCookie,
+                        bot.syncController.syncCookie,
                     ).sendAndExpect()
                 }
                 return
@@ -190,7 +181,7 @@ internal object MessageSvcPbGetMsg : OutgoingPacketFactory<MessageSvcPbGetMsg.Re
                     MessageSvcPbGetMsg(
                         client,
                         MsgSvc.SyncFlag.CONTINUE,
-                        bot.client.syncingController.syncCookie,
+                        bot.syncController.syncCookie,
                     ).sendAndExpect()
                 }
                 return

mirai-core/src/commonMain/kotlin/network/protocol/packet/chat/receive/MessageSvc.PbSendMsg.kt

@@ -23,7 +23,8 @@ import net.mamoe.mirai.internal.contact.uin
 import net.mamoe.mirai.internal.message.*
 import net.mamoe.mirai.internal.network.Packet
 import net.mamoe.mirai.internal.network.QQAndroidClient
-import net.mamoe.mirai.internal.network.QQAndroidClient.MessageSvcSyncData.PendingGroupMessageReceiptSyncId
+import net.mamoe.mirai.internal.network.components.SyncController.Companion.syncController
+import net.mamoe.mirai.internal.network.components.SyncController.Companion.syncCookie
 import net.mamoe.mirai.internal.network.protocol.data.proto.ImMsgBody
 import net.mamoe.mirai.internal.network.protocol.data.proto.MsgComm
 import net.mamoe.mirai.internal.network.protocol.data.proto.MsgCtrl
@@ -104,13 +105,13 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
             msgBody: ImMsgBody.MsgBody,
             msgSeq: Int,
             msgRand: Int,
-            contentHead: MsgComm.ContentHead
+            contentHead: MsgComm.ContentHead,
         ) -> MsgSvc.PbSendMsgReq,
         sequenceIds: AtomicReference<IntArray>,
         sequenceIdsInitializer: (Int) -> IntArray,
         randIds: AtomicReference<IntArray>,
         doFragmented: Boolean = true,
-        postInit: () -> Unit
+        postInit: () -> Unit,
     ): List<OutgoingPacket> {
         val fragmented = if (doFragmented)
             message.fragmented()
@@ -126,7 +127,8 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
         randIds.set(randIds0)
         postInit()
         fragmented.forEachIndexed { pkgIndex, fMsg ->
-            response.add(buildOutgoingUniPacket(client) {
+            response.add(
+                buildOutgoingUniPacket(client) {
                     writeProtoBuf(
                         MsgSvc.PbSendMsgReq.serializer(),
                         pbSendMsgReq(
@@ -136,11 +138,12 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                             MsgComm.ContentHead(
                                 pkgNum = pkgNum,
                                 divSeq = div,
-                            pkgIndex = pkgIndex
+                                pkgIndex = pkgIndex,
+                            ),
+                        ),
                     )
+                },
             )
-                )
-            })
         }
         return response
     }
@@ -162,7 +165,7 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                 } else {
                     0
                 }
-            }
+            },
         )
     }
 
@@ -175,7 +178,7 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
         target: Stranger,
         message: MessageChain,
         fragmented: Boolean,
-        source: (OnlineMessageSourceToStrangerImpl) -> Unit
+        source: (OnlineMessageSourceToStrangerImpl) -> Unit,
     ): List<OutgoingPacket> {
 
         val sequenceIds = AtomicReference<IntArray>()
@@ -186,8 +189,8 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
             fragmentTranslator = {
                 ImMsgBody.MsgBody(
                     richText = ImMsgBody.RichText(
-                        elems = it.toRichTextElems(messageTarget = target, withGeneralFlags = true)
+                        elems = it.toRichTextElems(messageTarget = target, withGeneralFlags = true),
-                    )
+                    ),
                 )
             },
             pbSendMsgReq = { msgBody, msgSeq, msgRand, contentHead ->
@@ -197,7 +200,7 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                     msgBody = msgBody,
                     msgSeq = msgSeq,
                     msgRand = msgRand,
-                    syncCookie = client.syncingController.syncCookie ?: byteArrayOf()
+                    syncCookie = client.syncCookie ?: byteArrayOf(),
                     // msgVia = 1
                 )
             },
@@ -214,11 +217,11 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                         target = target,
                         time = currentTimeSeconds().toInt(),
                         sequenceIds = sequenceIds.get(),
-                        originalMessage = message
+                        originalMessage = message,
-                    )
+                    ),
                 )
             },
-            doFragmented = fragmented
+            doFragmented = fragmented,
         )
     }
 
@@ -231,7 +234,7 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
         targetFriend: Friend,
         message: MessageChain,
         fragmented: Boolean,
-        crossinline sourceCallback: (OnlineMessageSourceToFriendImpl) -> Unit
+        crossinline sourceCallback: (OnlineMessageSourceToFriendImpl) -> Unit,
     ): List<OutgoingPacket> {
         contract {
             callsInPlace(sourceCallback, InvocationKind.EXACTLY_ONCE)
@@ -247,7 +250,7 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                     richText = ImMsgBody.RichText(
                         elems = subChain.toRichTextElems(messageTarget = targetFriend, withGeneralFlags = true),
                         ptt = subChain.findPtt(),
-                    )
+                    ),
                 )
             },
             pbSendMsgReq = { msgBody, msgSeq, msgRand, contentHead ->
@@ -257,7 +260,7 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                     msgBody = msgBody,
                     msgSeq = msgSeq,
                     msgRand = msgRand,
-                    syncCookie = client.syncingController.syncCookie ?: byteArrayOf()
+                    syncCookie = client.syncCookie ?: byteArrayOf(),
                     // msgVia = 1
                 )
             },
@@ -274,8 +277,8 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                         target = targetFriend,
                         time = currentTimeSeconds().toInt(),
                         sequenceIds = sequenceIds.get(),
-                        originalMessage = message
+                        originalMessage = message,
-                    )
+                    ),
                 )
             },
             doFragmented = fragmented,
@@ -307,7 +310,7 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                 ),
                 msgSeq = source.sequenceIds.single(),
                 msgRand = source.internalIds.single(),
-                syncCookie = client.syncingController.syncCookie ?: byteArrayOf()
+                syncCookie = client.syncCookie ?: byteArrayOf()
                 // msgVia = 1
             )
         )
@@ -322,23 +325,24 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
         client: QQAndroidClient,
         targetMember: Member,
         message: MessageChain,
-        source: OnlineMessageSourceToTempImpl
+        source: OnlineMessageSourceToTempImpl,
     ) = buildOutgoingUniPacket(client) {
         writeProtoBuf(
-            MsgSvc.PbSendMsgReq.serializer(), MsgSvc.PbSendMsgReq(
+            MsgSvc.PbSendMsgReq.serializer(),
+            MsgSvc.PbSendMsgReq(
                 routingHead = MsgSvc.RoutingHead(
-                    grpTmp = MsgSvc.GrpTmp(targetMember.group.uin, targetMember.id)
+                    grpTmp = MsgSvc.GrpTmp(targetMember.group.uin, targetMember.id),
                 ),
                 contentHead = MsgComm.ContentHead(pkgNum = 1),
                 msgBody = ImMsgBody.MsgBody(
                     richText = ImMsgBody.RichText(
-                        elems = message.toRichTextElems(messageTarget = targetMember, withGeneralFlags = true)
+                        elems = message.toRichTextElems(messageTarget = targetMember, withGeneralFlags = true),
-                    )
+                    ),
                 ),
                 msgSeq = source.sequenceIds.single(),
                 msgRand = source.internalIds.single(),
-                syncCookie = client.syncingController.syncCookie ?: byteArrayOf()
+                syncCookie = client.syncCookie ?: byteArrayOf(),
-            )
+            ),
         )
     }
 
@@ -352,7 +356,7 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
         targetGroup: Group,
         message: MessageChain,
         fragmented: Boolean,
-        crossinline sourceCallback: (OnlineMessageSourceToGroupImpl) -> Unit
+        crossinline sourceCallback: (OnlineMessageSourceToGroupImpl) -> Unit,
     ): List<OutgoingPacket> {
         val sequenceIds = AtomicReference<IntArray>()
         val randIds = AtomicReference<IntArray>()
@@ -363,9 +367,9 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                 ImMsgBody.MsgBody(
                     richText = ImMsgBody.RichText(
                         elems = subChain.toRichTextElems(messageTarget = targetGroup, withGeneralFlags = true),
-                        ptt = subChain.findPtt()
+                        ptt = subChain.findPtt(),
 
-                    )
+                        ),
                 )
             },
             pbSendMsgReq = { msgBody, msgSeq, msgRand, contentHead ->
@@ -375,12 +379,12 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                     msgBody = msgBody,
                     msgSeq = msgSeq,
                     msgRand = msgRand,
-                    syncCookie = client.syncingController.syncCookie ?: byteArrayOf(),
+                    syncCookie = client.syncCookie ?: byteArrayOf(),
                     msgVia = 1,
                     msgCtrl =
                     if (message[ForwardMessageInternal] != null)
                         MsgCtrl.MsgCtrl(msgFlag = 4)
-                    else null
+                    else null,
                 )
             },
             sequenceIds = sequenceIds,
@@ -389,13 +393,7 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                 IntArray(size) { client.atomicNextMessageSequenceId() }
             },
             postInit = {
-                randIds.get().forEach { id ->
+                randIds.get().forEach { id -> client.syncController.syncGroupMessageReceipt(id) }
-                    client.syncingController.pendingGroupMessageReceiptCacheList.addCache(
-                        PendingGroupMessageReceiptSyncId(
-                            messageRandom = id,
-                        )
-                    )
-                }
                 sourceCallback(
                     OnlineMessageSourceToGroupImpl(
                         targetGroup,
@@ -403,12 +401,12 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
                         sender = client.bot,
                         target = targetGroup,
                         time = currentTimeSeconds().toInt(),
-                        originalMessage = message//,
+                        originalMessage = message, //,
                         //   sourceMessage = message
-                    )
+                    ),
                 )
             },
-            doFragmented = fragmented
+            doFragmented = fragmented,
         )
     }
 
@@ -468,7 +466,7 @@ internal object MessageSvcPbSendMsg : OutgoingPacketFactory<MessageSvcPbSendMsg.
             else -> Response.Failed(
                 response.result,
                 response.errtype,
-                response.errmsg
+                response.errmsg,
             )
 
         }
@@ -481,7 +479,7 @@ internal inline fun MessageSvcPbSendMsg.createToTemp(
     member: Member,
     message: MessageChain,
     fragmented: Boolean,
-    crossinline sourceCallback: (Deferred<OnlineMessageSourceToTempImpl>) -> Unit
+    crossinline sourceCallback: (Deferred<OnlineMessageSourceToTempImpl>) -> Unit,
 ): List<OutgoingPacket> {
     contract {
         callsInPlace(sourceCallback, InvocationKind.EXACTLY_ONCE)
@@ -492,14 +490,14 @@ internal inline fun MessageSvcPbSendMsg.createToTemp(
         target = member,
         time = currentTimeSeconds().toInt(),
         sequenceIds = intArrayOf(client.atomicNextMessageSequenceId()),
-        originalMessage = message
+        originalMessage = message,
     )
     sourceCallback(CompletableDeferred(source))
     return createToTempImpl(
         client,
         member,
         message,
-        source
+        source,
     ).let { listOf(it) }
 }
 
@@ -508,7 +506,7 @@ internal inline fun MessageSvcPbSendMsg.createToStranger(
     stranger: Stranger,
     message: MessageChain,
     fragmented: Boolean,
-    crossinline sourceCallback: (Deferred<OnlineMessageSourceToStrangerImpl>) -> Unit
+    crossinline sourceCallback: (Deferred<OnlineMessageSourceToStrangerImpl>) -> Unit,
 ): List<OutgoingPacket> {
     contract {
         callsInPlace(sourceCallback, InvocationKind.EXACTLY_ONCE)
@@ -517,7 +515,7 @@ internal inline fun MessageSvcPbSendMsg.createToStranger(
         client,
         stranger,
         message,
-        fragmented
+        fragmented,
     ) { sourceCallback(CompletableDeferred(it)) }
 }
 
@@ -526,7 +524,7 @@ internal inline fun MessageSvcPbSendMsg.createToFriend(
     qq: Friend,
     message: MessageChain,
     fragmented: Boolean,
-    crossinline sourceCallback: (Deferred<OnlineMessageSourceToFriendImpl>) -> Unit
+    crossinline sourceCallback: (Deferred<OnlineMessageSourceToFriendImpl>) -> Unit,
 ): List<OutgoingPacket> {
     contract {
         callsInPlace(sourceCallback, InvocationKind.EXACTLY_ONCE)
@@ -535,7 +533,7 @@ internal inline fun MessageSvcPbSendMsg.createToFriend(
         client,
         qq,
         message,
-        fragmented
+        fragmented,
     ) { sourceCallback(CompletableDeferred(it)) }
 }
 
@@ -545,7 +543,7 @@ internal inline fun MessageSvcPbSendMsg.createToGroup(
     group: Group,
     message: MessageChain,
     fragmented: Boolean,
-    crossinline sourceCallback: (Deferred<OnlineMessageSourceToGroupImpl>) -> Unit
+    crossinline sourceCallback: (Deferred<OnlineMessageSourceToGroupImpl>) -> Unit,
 ): List<OutgoingPacket> {
     contract {
         callsInPlace(sourceCallback, InvocationKind.EXACTLY_ONCE)
@@ -554,6 +552,6 @@ internal inline fun MessageSvcPbSendMsg.createToGroup(
         client,
         group,
         message,
-        fragmented
+        fragmented,
     ) { sourceCallback(CompletableDeferred(it)) }
 }

mirai-core/src/commonMain/kotlin/network/protocol/packet/chat/receive/MessageSvc.PushNotify.kt

@@ -13,6 +13,7 @@ import kotlinx.atomicfu.loop
 import kotlinx.io.core.ByteReadPacket
 import kotlinx.io.core.discardExact
 import net.mamoe.mirai.internal.QQAndroidBot
+import net.mamoe.mirai.internal.network.components.SyncController.Companion.syncController
 import net.mamoe.mirai.internal.network.protocol.data.jce.RequestPushNotify
 import net.mamoe.mirai.internal.network.protocol.data.proto.MsgSvc
 import net.mamoe.mirai.internal.network.protocol.packet.IncomingPacketFactory
@@ -30,18 +31,17 @@ internal object MessageSvcPushNotify : IncomingPacketFactory<RequestPushNotify>(
     }
 
     override suspend fun QQAndroidBot.handle(packet: RequestPushNotify, sequenceId: Int): OutgoingPacket {
-
+        syncController.firstNotify.loop { firstNotify ->
-        client.syncingController.firstNotify.loop { firstNotify ->
             network.run {
                 return MessageSvcPbGetMsg(
                     client,
                     MsgSvc.SyncFlag.START,
                     if (firstNotify) {
-                        if (!client.syncingController.firstNotify.compareAndSet(firstNotify, false)) {
+                        if (!syncController.firstNotify.compareAndSet(firstNotify, false)) {
                             return@loop
                         }
                         null
-                    } else packet.vNotifyCookie
+                    } else packet.vNotifyCookie,
                 )
             }
         }

mirai-core/src/commonMain/kotlin/network/protocol/packet/chat/receive/MessageSvc.RequestPushStatus.kt

@@ -12,6 +12,7 @@ package net.mamoe.mirai.internal.network.protocol.packet.chat.receive
 import kotlinx.io.core.ByteReadPacket
 import net.mamoe.mirai.internal.QQAndroidBot
 import net.mamoe.mirai.internal.network.Packet
+import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.processPacketThroughPipeline
 import net.mamoe.mirai.internal.network.protocol.data.jce.RequestPushStatus
 import net.mamoe.mirai.internal.network.protocol.packet.IncomingPacketFactory
 import net.mamoe.mirai.internal.utils.io.serialization.readUniPacket

mirai-core/src/commonMain/kotlin/network/protocol/packet/chat/receive/OnlinePush.PbC2CMsgSync.kt

@@ -12,21 +12,19 @@ package net.mamoe.mirai.internal.network.protocol.packet.chat.receive
 import kotlinx.io.core.ByteReadPacket
 import net.mamoe.mirai.internal.QQAndroidBot
 import net.mamoe.mirai.internal.network.Packet
-import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.noticeProcessorPipeline
+import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.processPacketThroughPipeline
 import net.mamoe.mirai.internal.network.components.PipelineContext.Companion.KEY_FROM_SYNC
 import net.mamoe.mirai.internal.network.protocol.data.proto.MsgOnlinePush
 import net.mamoe.mirai.internal.network.protocol.packet.IncomingPacketFactory
-import net.mamoe.mirai.internal.network.toPacket
 import net.mamoe.mirai.internal.utils.io.serialization.readProtoBuf
 
 internal object PbC2CMsgSync : IncomingPacketFactory<Packet>(
-    "OnlinePush.PbC2CMsgSync", ""
+    "OnlinePush.PbC2CMsgSync", "",
 ) {
     override suspend fun ByteReadPacket.decode(bot: QQAndroidBot, sequenceId: Int): Packet {
-        return bot.components.noticeProcessorPipeline.process(
+        return bot.processPacketThroughPipeline(
-            bot = bot,
+            readProtoBuf(MsgOnlinePush.PbPushMsg.serializer()).msg,
-            data = readProtoBuf(MsgOnlinePush.PbPushMsg.serializer()).msg,
+            KEY_FROM_SYNC to true,
-            attributes = KEY_FROM_SYNC to true,
+        )
-        ).toPacket()
     }
 }

mirai-core/src/commonMain/kotlin/network/protocol/packet/chat/receive/OnlinePush.PbPushGroupMsg.kt

@@ -7,18 +7,15 @@
  *  https://github.com/mamoe/mirai/blob/master/LICENSE
  */
 
-@file:Suppress("INVISIBLE_MEMBER", "INVISIBLE_REFERENCE")
-
 package net.mamoe.mirai.internal.network.protocol.packet.chat.receive
 
 import kotlinx.io.core.ByteReadPacket
 import net.mamoe.mirai.internal.QQAndroidBot
 import net.mamoe.mirai.internal.network.Packet
-import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.noticeProcessorPipeline
+import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.processPacketThroughPipeline
 import net.mamoe.mirai.internal.network.components.SsoProcessor
 import net.mamoe.mirai.internal.network.protocol.data.proto.MsgOnlinePush
 import net.mamoe.mirai.internal.network.protocol.packet.IncomingPacketFactory
-import net.mamoe.mirai.internal.network.toPacket
 import net.mamoe.mirai.internal.utils.io.serialization.readProtoBuf
 
 /**
@@ -26,11 +23,9 @@ import net.mamoe.mirai.internal.utils.io.serialization.readProtoBuf
  */
 internal object OnlinePushPbPushGroupMsg : IncomingPacketFactory<Packet?>("OnlinePush.PbPushGroupMsg") {
 
-    @OptIn(ExperimentalStdlibApi::class)
     override suspend fun ByteReadPacket.decode(bot: QQAndroidBot, sequenceId: Int): Packet? {
         // 00 00 02 E4 0A D5 05 0A 4F 08 A2 FF 8C F0 03 10 DD F1 92 B7 07 18 52 20 00 28 BC 3D 30 8C 82 AB F1 05 38 D2 80 E0 8C 80 80 80 80 02 4A 21 08 E7 C1 AD B8 02 10 01 18 BA 05 22 09 48 69 6D 31 38 38 6D 6F 65 30 06 38 02 42 05 4D 69 72 61 69 50 01 58 01 60 00 88 01 08 12 06 08 01 10 00 18 00 1A F9 04 0A F6 04 0A 26 08 00 10 87 82 AB F1 05 18 B7 B4 BF 30 20 00 28 0C 30 00 38 86 01 40 22 4A 0C E5 BE AE E8 BD AF E9 9B 85 E9 BB 91 12 E6 03 42 E3 03 12 2A 7B 34 45 31 38 35 38 32 32 2D 30 45 37 42 2D 46 38 30 46 2D 43 35 42 31 2D 33 34 34 38 38 33 37 34 44 33 39 43 7D 2E 6A 70 67 22 00 2A 04 03 00 00 00 32 60 15 36 20 39 36 6B 45 31 41 38 35 32 32 39 64 63 36 39 38 34 37 39 37 37 62 20 20 20 20 20 20 35 30 20 20 20 20 20 20 20 20 20 20 20 20 20 20 20 20 7B 34 45 31 38 35 38 32 32 2D 30 45 37 42 2D 46 38 30 46 2D 43 35 42 31 2D 33 34 34 38 38 33 37 34 44 33 39 43 7D 2E 6A 70 67 31 32 31 32 41 38 C6 BB 8A A9 08 40 FB AE 9E C2 09 48 50 50 41 5A 00 60 01 6A 10 4E 18 58 22 0E 7B F8 0F C5 B1 34 48 83 74 D3 9C 72 59 2F 67 63 68 61 74 70 69 63 5F 6E 65 77 2F 31 30 34 30 34 30 30 32 39 30 2F 36 35 35 30 35 37 31 32 37 2D 32 32 33 33 36 33 38 33 34 32 2D 34 45 31 38 35 38 32 32 30 45 37 42 46 38 30 46 43 35 42 31 33 34 34 38 38 33 37 34 44 33 39 43 2F 31 39 38 3F 74 65 72 6D 3D 32 82 01 57 2F 67 63 68 61 74 70 69 63 5F 6E 65 77 2F 31 30 34 30 34 30 30 32 39 30 2F 36 35 35 30 35 37 31 32 37 2D 32 32 33 33 36 33 38 33 34 32 2D 34 45 31 38 35 38 32 32 30 45 37 42 46 38 30 46 43 35 42 31 33 34 34 38 38 33 37 34 44 33 39 43 2F 30 3F 74 65 72 6D 3D 32 B0 01 4D B8 01 2E C8 01 FF 05 D8 01 4D E0 01 2E FA 01 59 2F 67 63 68 61 74 70 69 63 5F 6E 65 77 2F 31 30 34 30 34 30 30 32 39 30 2F 36 35 35 30 35 37 31 32 37 2D 32 32 33 33 36 33 38 33 34 32 2D 34 45 31 38 35 38 32 32 30 45 37 42 46 38 30 46 43 35 42 31 33 34 34 38 38 33 37 34 44 33 39 43 2F 34 30 30 3F 74 65 72 6D 3D 32 80 02 4D 88 02 2E 12 45 AA 02 42 50 03 60 00 68 00 9A 01 39 08 09 20 BF 50 80 01 01 C8 01 00 F0 01 00 F8 01 00 90 02 00 98 03 00 A0 03 20 B0 03 00 C0 03 00 D0 03 00 E8 03 00 8A 04 04 08 02 08 01 90 04 80 80 80 10 B8 04 00 C0 04 00 12 06 4A 04 08 00 40 01 12 14 82 01 11 0A 09 48 69 6D 31 38 38 6D 6F 65 18 06 20 08 28 03 10 8A CA 9D A1 07 1A 00
         if (!bot.components[SsoProcessor].firstLoginSucceed) return null
-        return bot.components.noticeProcessorPipeline.process(bot, readProtoBuf(MsgOnlinePush.PbPushMsg.serializer()))
+        return bot.processPacketThroughPipeline(readProtoBuf(MsgOnlinePush.PbPushMsg.serializer()))
-            .toPacket()
     }
 }

mirai-core/src/commonMain/kotlin/network/protocol/packet/chat/receive/OnlinePush.PbPushTransMsg.kt

@@ -12,8 +12,9 @@ package net.mamoe.mirai.internal.network.protocol.packet.chat.receive
 import kotlinx.io.core.ByteReadPacket
 import net.mamoe.mirai.internal.QQAndroidBot
 import net.mamoe.mirai.internal.network.Packet
-import net.mamoe.mirai.internal.network.QQAndroidClient
+import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.processPacketThroughPipeline
-import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.noticeProcessorPipeline
+import net.mamoe.mirai.internal.network.components.SyncController.Companion.syncController
+import net.mamoe.mirai.internal.network.components.syncPushTrans
 import net.mamoe.mirai.internal.network.protocol.data.proto.OnlinePushTrans
 import net.mamoe.mirai.internal.network.protocol.packet.IncomingPacketFactory
 import net.mamoe.mirai.internal.network.protocol.packet.OutgoingPacket
@@ -27,20 +28,8 @@ internal object OnlinePushPbPushTransMsg :
 
     override suspend fun ByteReadPacket.decode(bot: QQAndroidBot, sequenceId: Int): Packet? {
         val content = this.readProtoBuf(OnlinePushTrans.PbMsgInfo.serializer())
-
+        if (!bot.syncController.syncPushTrans(content)) return null
-        if (!bot.client.syncingController.pbPushTransMsgCacheList.addCache(
+        return bot.processPacketThroughPipeline(content)
-                content.run {
-                    QQAndroidClient.MessageSvcSyncData.PbPushTransMsgSyncId(msgUid, msgSeq, msgTime)
-                }
-            )
-        ) {
-            return null
-        }
-        // bot.network.logger.debug { content._miraiContentToString() }
-
-
-        bot.components.noticeProcessorPipeline.process(bot, content)
-        return null
     }
 
     override suspend fun QQAndroidBot.handle(packet: Packet?, sequenceId: Int): OutgoingPacket {

mirai-core/src/commonMain/kotlin/network/protocol/packet/chat/receive/OnlinePush.ReqPush.kt

@@ -7,14 +7,12 @@
  *  https://github.com/mamoe/mirai/blob/master/LICENSE
  */
 
-@file:Suppress("INVISIBLE_MEMBER", "INVISIBLE_REFERENCE")
-@file:OptIn(
-    JavaFriendlyAPI::class
-)
-
 package net.mamoe.mirai.internal.network.protocol.packet.chat.receive
 
-import kotlinx.io.core.*
+import kotlinx.io.core.ByteReadPacket
+import kotlinx.io.core.discardExact
+import kotlinx.io.core.readUInt
+import kotlinx.io.core.readUShort
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.protobuf.ProtoNumber
 import net.mamoe.mirai.contact.Member
@@ -26,14 +24,13 @@ import net.mamoe.mirai.internal.QQAndroidBot
 import net.mamoe.mirai.internal.contact.GroupImpl
 import net.mamoe.mirai.internal.contact.checkIsGroupImpl
 import net.mamoe.mirai.internal.contact.checkIsMemberImpl
-import net.mamoe.mirai.internal.network.MultiPacketImpl
+import net.mamoe.mirai.internal.network.MultiPacket
 import net.mamoe.mirai.internal.network.Packet
-import net.mamoe.mirai.internal.network.QQAndroidClient
+import net.mamoe.mirai.internal.network.components.NoticeProcessorPipeline.Companion.processPacketThroughPipeline
 import net.mamoe.mirai.internal.network.handler.logger
 import net.mamoe.mirai.internal.network.protocol.data.jce.MsgInfo
 import net.mamoe.mirai.internal.network.protocol.data.jce.MsgType0x210
 import net.mamoe.mirai.internal.network.protocol.data.jce.OnlinePushPack
-import net.mamoe.mirai.internal.network.protocol.data.jce.RequestPacket
 import net.mamoe.mirai.internal.network.protocol.data.proto.Submsgtype0x122
 import net.mamoe.mirai.internal.network.protocol.data.proto.Submsgtype0x27.SubMsgType0x27.*
 import net.mamoe.mirai.internal.network.protocol.data.proto.TroopTips0x857
@@ -42,112 +39,49 @@ import net.mamoe.mirai.internal.network.protocol.packet.OutgoingPacket
 import net.mamoe.mirai.internal.network.protocol.packet.buildResponseUniPacket
 import net.mamoe.mirai.internal.utils._miraiContentToString
 import net.mamoe.mirai.internal.utils.io.ProtoBuf
-import net.mamoe.mirai.internal.utils.io.serialization.*
+import net.mamoe.mirai.internal.utils.io.serialization.loadAs
-import net.mamoe.mirai.utils.*
+import net.mamoe.mirai.internal.utils.io.serialization.readProtoBuf
+import net.mamoe.mirai.internal.utils.io.serialization.readUniPacket
+import net.mamoe.mirai.internal.utils.io.serialization.writeJceRequestPacket
+import net.mamoe.mirai.utils.currentTimeSeconds
+import net.mamoe.mirai.utils.debug
+import net.mamoe.mirai.utils.encodeToString
+import net.mamoe.mirai.utils.mapToIntArray
 
 
 //0C 01 B1 89 BE 09 5E 3D 72 A6 00 01 73 68 FC 06 00 00 00 3C
 internal object OnlinePushReqPush : IncomingPacketFactory<OnlinePushReqPush.ReqPushDecoded>(
     "OnlinePush.ReqPush",
-    "OnlinePush.RespPush"
+    "OnlinePush.RespPush",
 ) {
-    // to reduce nesting depth
-    private suspend fun List<MsgInfo>.deco(
-        client: QQAndroidClient,
-        mapper: suspend ByteReadPacket.(msgInfo: MsgInfo) -> Sequence<Packet>
-    ): List<Packet> {
-        return mapNotNull { msg ->
-            val successful = client.syncingController.onlinePushReqPushCacheList.addCache(
-                QQAndroidClient.MessageSvcSyncData.OnlinePushReqPushSyncId(
-                    uid = msg.lMsgUid ?: 0, sequence = msg.shMsgSeq, time = msg.uMsgTime
-                )
-            )
-            if (!successful) return@mapNotNull null
-            msg.vMsg.read { mapper(msg) }
-        }.asSequence().flatten().toList()
-    }
-
-
-    @ExperimentalUnsignedTypes
-    @OptIn(ExperimentalStdlibApi::class)
     override suspend fun ByteReadPacket.decode(bot: QQAndroidBot, sequenceId: Int): ReqPushDecoded {
         val reqPushMsg = readUniPacket(OnlinePushPack.SvcReqPushMsg.serializer(), "req")
-        //bot.network.logger.debug { reqPushMsg._miraiContentToString() }
+        return ReqPushDecoded(reqPushMsg, bot.processPacketThroughPipeline(reqPushMsg))
-        val packets = reqPushMsg.vMsgInfos.deco(bot.client) { msgInfo ->
-            when (msgInfo.shMsgType.toInt()) {
-                732 -> {
-                    val group = bot.getGroup(readUInt().toLong())
-                        ?: return@deco emptySequence() // group has not been initialized
-
-                    group.checkIsGroupImpl()
-
-                    val internalType = readByte().toInt()
-                    discardExact(1)
-
-                    Transformers732[internalType]
-                        ?.let { it(this@deco, group, bot) }
-                        ?: kotlin.run {
-                            bot.network.logger.debug {
-                                "unknown group 732 type $internalType, data: " + readBytes().toUHexString()
-                            }
-                            return@deco emptySequence()
-                        }
     }
 
-                // 00 27 1A 0C 1C 2C 3C 4C 5D 00 0C 6D 00 0C 7D 00 0C 8D 00 0C 9C AC BC CC DD 00 0C EC FC 0F 0B 2A 0C 1C 2C 3C 4C 5C 6C 0B 3A 0C 1C 2C 3C 4C 5C 6C 7C 8D 00 0C 9D 00 0C AC BD 00 0C CD 00 0C DC ED 00 0C FC 0F FC 10 0B 4A 0C 1C 2C 3C 4C 5C 6C 7C 8C 96 00 0B 5A 0C 1C 2C 3C 4C 5C 6C 7C 8C 9D 00 0C 0B 6A 0C 1A 0C 1C 26 00 0B 2A 0C 0B 3A 0C 16 00 0B 4A 09 0C 0B 5A 09 0C 0B 0B 7A 0C 1C 2C 36 00 0B 8A 0C 1C 2C 36 00 0B 9A 09 0C 0B AD 00 00 1E 0A 1C 10 28 4A 18 0A 16 08 00 10 A2 FF 8C F0 03 1A 0C E6 BD 9C E6 B1 9F E7 BE A4 E5 8F 8B
+    internal class ReqPushDecoded(val request: OnlinePushPack.SvcReqPushMsg, packet: Packet) :
-                528 -> {
+        MultiPacket by MultiPacket(packet), Packet.NoLog {
-                    val notifyMsgBody = readJceStruct(MsgType0x210.serializer())
+        override fun toString(): String = "OnlinePush.ReqPush.ReqPushDecoded"
-                    Transformers528[notifyMsgBody.uSubMsgType]
-                        ?.let { processor -> processor(notifyMsgBody, bot, msgInfo) }
-                        ?: kotlin.run {
-                            bot.network.logger.debug {
-                                "unknown group 528 type 0x${notifyMsgBody.uSubMsgType.toUHexString("")}, data: " + notifyMsgBody.vProtobuf.toUHexString()
-                            }
-                            return@deco emptySequence()
-                        }
-                }
-                else -> {
-                    bot.network.logger.debug { "unknown sh type ${msgInfo.shMsgType.toInt()}" }
-                    bot.network.logger.debug { "data=${readBytes().toUHexString()}" }
-                    return@deco emptySequence()
-                }
-            }
-        }
-        return ReqPushDecoded(reqPushMsg, packets)
-    }
-
-    @Suppress("SpellCheckingInspection")
-    internal data class ReqPushDecoded(val request: OnlinePushPack.SvcReqPushMsg, val sequence: Collection<Packet>) :
-        MultiPacketImpl(sequence), Packet.NoLog {
-        override fun toString(): String {
-            return "OnlinePush.ReqPush.ReqPushDecoded"
-        }
     }
 
     override suspend fun QQAndroidBot.handle(packet: ReqPushDecoded, sequenceId: Int): OutgoingPacket {
         return buildResponseUniPacket(client) {
-            writeJceStruct(
+            writeJceRequestPacket(
-                RequestPacket.serializer(),
-                RequestPacket(
                 servantName = "OnlinePush",
                 funcName = "SvcRespPushMsg",
-                    requestId = sequenceId,
+                name = "resp",
-                    sBuffer = jceRequestSBuffer(
+                serializer = OnlinePushPack.SvcRespPushMsg.serializer(),
-                        "resp",
+                body = OnlinePushPack.SvcRespPushMsg(
-                        OnlinePushPack.SvcRespPushMsg.serializer(),
-                        OnlinePushPack.SvcRespPushMsg(
                     packet.request.uin,
                     packet.request.vMsgInfos.map { msg ->
                         OnlinePushPack.DelMsgInfo(
                             fromUin = msg.lFromUin,
                             shMsgSeq = msg.shMsgSeq,
                             vMsgCookies = msg.vMsgCookies,
-                                    uMsgTime = msg.uMsgTime // captured 0
+                            uMsgTime = msg.uMsgTime, // captured 0
-                                )
-                            }
-                        )
-                    )
                         )
+                    },
+                ),
             )
         }
     }
@@ -171,7 +105,7 @@ private fun handleMuteMemberPacket(
     group: GroupImpl,
     operator: NormalMember,
     target: Long,
-    timeSeconds: Int
+    timeSeconds: Int,
 ): Packet? {
     if (target == 0L) {
         val new = timeSeconds != 0
@@ -209,7 +143,7 @@ private fun handleMuteMemberPacket(
     else MemberMuteEvent(member, timeSeconds, operator)
 }
 
-private object Transformers732 : Map<Int, Lambda732> by mapOf(
+internal object Transformers732 : Map<Int, Lambda732> by mapOf(
     // mute
     0x0c to lambda732 { group: GroupImpl, bot: QQAndroidBot ->
         val operatorUin = readUInt().toLong()
@@ -274,8 +208,8 @@ private object Transformers732 : Map<Int, Lambda732> by mapOf(
                         target = if (target.id == bot.id) bot else target,
                         action = action,
                         suffix = suffix,
-                        subject = group
+                        subject = group,
-                    )
+                    ),
                 )
             }
             //龙王
@@ -294,7 +228,7 @@ private object Transformers732 : Map<Int, Lambda732> by mapOf(
                     sequenceOf(
                         GroupTalkativeChangeEvent(group, now, it),
                         MemberHonorChangeEvent.Lose(it, GroupHonorType.TALKATIVE),
-                        MemberHonorChangeEvent.Achieve(now, GroupHonorType.TALKATIVE)
+                        MemberHonorChangeEvent.Achieve(now, GroupHonorType.TALKATIVE),
                     )
                 } ?: sequenceOf(MemberHonorChangeEvent.Achieve(now, GroupHonorType.TALKATIVE))
             }
@@ -331,8 +265,8 @@ private object Transformers732 : Map<Int, Lambda732> by mapOf(
                                     new,
                                     !new,
                                     group,
-                                    false
+                                    false,
-                                )
+                                ),
                             )
                         }
                         else -> {
@@ -439,18 +373,17 @@ private object Transformers732 : Map<Int, Lambda732> by mapOf(
                     firstPkg.time,
                     operator,
                     group,
-                    group[firstPkg.authorUin] ?: return@lambda732 emptySequence()
+                    group[firstPkg.authorUin] ?: return@lambda732 emptySequence(),
-                )
+                ),
             )
         }
-    }
+    },
 )
 
 internal interface Lambda528 {
     suspend operator fun invoke(msg: MsgType0x210, bot: QQAndroidBot, msgInfo: MsgInfo): Sequence<Packet>
 }
 
-@kotlin.internal.LowPriorityInOverloadResolution
 internal inline fun lambda528(crossinline block: suspend MsgType0x210.(QQAndroidBot) -> Sequence<Packet>): Lambda528 {
     return object : Lambda528 {
         override suspend fun invoke(msg: MsgType0x210, bot: QQAndroidBot, msgInfo: MsgInfo): Sequence<Packet> {
@@ -472,7 +405,7 @@ internal inline fun lambda528(crossinline block: suspend MsgType0x210.(QQAndroid
 @Serializable
 private class Wording(
     @ProtoNumber(1) val itemID: Int = 0,
-    @ProtoNumber(2) val itemName: String = ""
+    @ProtoNumber(2) val itemName: String = "",
 ) : ProtoBuf
 
 @Serializable
@@ -487,7 +420,7 @@ private class Sub8AMsgInfo(
     @ProtoNumber(8) val pkgIndex: Int, // 0
     @ProtoNumber(9) val devSeq: Int, // 0
     @ProtoNumber(12) val flag: Int, // 1
-    @ProtoNumber(13) val wording: Wording
+    @ProtoNumber(13) val wording: Wording,
 ) : ProtoBuf
 
 @Serializable
@@ -496,7 +429,7 @@ private class Sub8A(
     @ProtoNumber(2) val appId: Int, // 1
     @ProtoNumber(3) val instId: Int, // 1
     @ProtoNumber(4) val longMessageFlag: Int, // 0
-    @ProtoNumber(5) val reserved: ByteArray? = null // struct{ boolean(1), boolean(2) }
+    @ProtoNumber(5) val reserved: ByteArray? = null, // struct{ boolean(1), boolean(2) }
 ) : ProtoBuf
 
 
@@ -519,7 +452,7 @@ internal object Transformers528 : Map<Long, Lambda528> by mapOf(
                     messageInternalIds = intArrayOf(info.srcInternalId.toInt()),
                     messageTime = info.time.toInt(),
                     operatorId = info.fromUin,
-                    operator = bot.getFriend(info.fromUin) ?: return@mapNotNull null
+                    operator = bot.getFriend(info.fromUin) ?: return@mapNotNull null,
                 )
             }
     },
@@ -567,7 +500,7 @@ internal object Transformers528 : Map<Long, Lambda528> by mapOf(
                             NudgeEvent(from = bot, target = subject, subject = subject, action, suffix)
                         }
                         else -> NudgeEvent(from = subject, target = subject, subject = subject, action, suffix)
-                    }
+                    },
                 )
             }
             else -> {

mirai-core/src/commonMain/kotlin/network/protocol/packet/chat/receive/OnlinePush.SidExpired.kt

@@ -22,14 +22,9 @@ import net.mamoe.mirai.internal.network.protocol.packet.sendAndExpect
 internal object OnlinePushSidExpired : IncomingPacketFactory<Packet?>("OnlinePush.SidTicketExpired") {
 
     override suspend fun QQAndroidBot.handle(packet: Packet?, sequenceId: Int): OutgoingPacket {
-        return buildResponseUniPacket(
-            client,
-            sequenceId = sequenceId,
-            key = client.wLoginSigInfo.d2Key
-        ) {}.also {
         WtLogin10(client, mainSigMap = 3554528).sendAndExpect(bot)
         StatSvc.Register.online(client).sendAndExpect(bot)
-        }
+        return buildResponseUniPacket(client, sequenceId = sequenceId)
     }
 
     override suspend fun ByteReadPacket.decode(bot: QQAndroidBot, sequenceId: Int): Packet? {

mirai-core/src/commonMain/kotlin/utils/io/ProtoBuf.kt

@@ -1,15 +0,0 @@
-/*
- * Copyright 2019-2021 Mamoe Technologies and contributors.
- *
- * 此源代码的使用受 GNU AFFERO GENERAL PUBLIC LICENSE version 3 许可证的约束, 可以在以下链接找到该许可证.
- * Use of this source code is governed by the GNU AGPLv3 license that can be found through the following link.
- *
- * https://github.com/mamoe/mirai/blob/dev/LICENSE
- */
-
-package net.mamoe.mirai.internal.utils.io
-
-/**
- * 仅有标示作用
- */
-internal interface ProtoBuf

mirai-core/src/commonMain/kotlin/utils/io/ProtocolStruct.kt

@@ -9,4 +9,6 @@
 
 package net.mamoe.mirai.internal.utils.io
 
-internal interface JceStruct
+internal interface ProtocolStruct
+internal interface ProtoBuf : ProtocolStruct
+internal interface JceStruct : ProtocolStruct