diff --git a/pkg/msgprocessor/conversation.go b/pkg/msgprocessor/conversation.go
new file mode 100644
index 000000000..b24794af8
--- /dev/null
+++ b/pkg/msgprocessor/conversation.go
@@ -0,0 +1,158 @@
+package msgprocessor
+
+import (
+	"github.com/OpenIMSDK/protocol/constant"
+	"github.com/OpenIMSDK/protocol/sdkws"
+	"google.golang.org/protobuf/proto"
+	"sort"
+	"strings"
+)
+
+func GetNotificationConversationID(msg *sdkws.MsgData) string {
+	switch msg.SessionType {
+	case constant.SingleChatType:
+		l := []string{msg.SendID, msg.RecvID}
+		sort.Strings(l)
+		return "n_" + strings.Join(l, "_")
+	case constant.GroupChatType:
+		return "n_" + msg.GroupID
+	case constant.SuperGroupChatType:
+		return "n_" + msg.GroupID
+	case constant.NotificationChatType:
+		return "n_" + msg.SendID + "_" + msg.RecvID
+	}
+	return ""
+}
+
+func GetChatConversationIDByMsg(msg *sdkws.MsgData) string {
+	switch msg.SessionType {
+	case constant.SingleChatType:
+		l := []string{msg.SendID, msg.RecvID}
+		sort.Strings(l)
+		return "si_" + strings.Join(l, "_")
+	case constant.GroupChatType:
+		return "g_" + msg.GroupID
+	case constant.SuperGroupChatType:
+		return "sg_" + msg.GroupID
+	case constant.NotificationChatType:
+		return "sn_" + msg.SendID + "_" + msg.RecvID
+	}
+	return ""
+}
+
+func GenConversationUniqueKey(msg *sdkws.MsgData) string {
+	switch msg.SessionType {
+	case constant.SingleChatType, constant.NotificationChatType:
+		l := []string{msg.SendID, msg.RecvID}
+		sort.Strings(l)
+		return strings.Join(l, "_")
+	case constant.SuperGroupChatType:
+		return msg.GroupID
+	}
+	return ""
+}
+
+func GetConversationIDByMsg(msg *sdkws.MsgData) string {
+	options := Options(msg.Options)
+	switch msg.SessionType {
+	case constant.SingleChatType:
+		l := []string{msg.SendID, msg.RecvID}
+		sort.Strings(l)
+		if !options.IsNotNotification() {
+			return "n_" + strings.Join(l, "_")
+		}
+		return "si_" + strings.Join(l, "_") // single chat
+	case constant.GroupChatType:
+		if !options.IsNotNotification() {
+			return "n_" + msg.GroupID // group chat
+		}
+		return "g_" + msg.GroupID // group chat
+	case constant.SuperGroupChatType:
+		if !options.IsNotNotification() {
+			return "n_" + msg.GroupID // super group chat
+		}
+		return "sg_" + msg.GroupID // super group chat
+	case constant.NotificationChatType:
+		if !options.IsNotNotification() {
+			return "n_" + msg.SendID + "_" + msg.RecvID // super group chat
+		}
+		return "sn_" + msg.SendID + "_" + msg.RecvID // server notification chat
+	}
+	return ""
+}
+
+func GetConversationIDBySessionType(sessionType int, ids ...string) string {
+	sort.Strings(ids)
+	if len(ids) > 2 || len(ids) < 1 {
+		return ""
+	}
+	switch sessionType {
+	case constant.SingleChatType:
+		return "si_" + strings.Join(ids, "_") // single chat
+	case constant.GroupChatType:
+		return "g_" + ids[0] // group chat
+	case constant.SuperGroupChatType:
+		return "sg_" + ids[0] // super group chat
+	case constant.NotificationChatType:
+		return "sn_" + ids[0] // server notification chat
+	}
+	return ""
+}
+
+func IsNotification(conversationID string) bool {
+	return strings.HasPrefix(conversationID, "n_")
+}
+
+func IsNotificationByMsg(msg *sdkws.MsgData) bool {
+	return !Options(msg.Options).IsNotNotification()
+}
+
+func ParseConversationID(msg *sdkws.MsgData) (isNotification bool, conversationID string) {
+	options := Options(msg.Options)
+	switch msg.SessionType {
+	case constant.SingleChatType:
+		l := []string{msg.SendID, msg.RecvID}
+		sort.Strings(l)
+		if !options.IsNotNotification() {
+			return true, "n_" + strings.Join(l, "_")
+		}
+		return false, "si_" + strings.Join(l, "_") // single chat
+	case constant.SuperGroupChatType:
+		if !options.IsNotNotification() {
+			return true, "n_" + msg.GroupID // super group chat
+		}
+		return false, "sg_" + msg.GroupID // super group chat
+	case constant.NotificationChatType:
+		if !options.IsNotNotification() {
+			return true, "n_" + msg.SendID + "_" + msg.RecvID // super group chat
+		}
+		return false, "sn_" + msg.SendID + "_" + msg.RecvID // server notification chat
+	}
+	return false, ""
+}
+
+type MsgBySeq []*sdkws.MsgData
+
+func (s MsgBySeq) Len() int {
+	return len(s)
+}
+
+func (s MsgBySeq) Less(i, j int) bool {
+	return s[i].Seq < s[j].Seq
+}
+
+func (s MsgBySeq) Swap(i, j int) {
+	s[i], s[j] = s[j], s[i]
+}
+
+func Pb2String(pb proto.Message) (string, error) {
+	s, err := proto.Marshal(pb)
+	if err != nil {
+		return "", err
+	}
+	return string(s), nil
+}
+
+func String2Pb(s string, pb proto.Message) error {
+	return proto.Unmarshal([]byte(s), pb)
+}
diff --git a/pkg/msgprocessor/options.go b/pkg/msgprocessor/options.go
new file mode 100644
index 000000000..920a6a2ea
--- /dev/null
+++ b/pkg/msgprocessor/options.go
@@ -0,0 +1,181 @@
+// Copyright © 2023 OpenIM. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package msgprocessor
+
+import "github.com/OpenIMSDK/protocol/constant"
+
+type Options map[string]bool
+type OptionsOpt func(Options)
+
+func NewOptions(opts ...OptionsOpt) Options {
+	options := make(map[string]bool, 11)
+	options[constant.IsNotNotification] = false
+	options[constant.IsSendMsg] = false
+	options[constant.IsHistory] = false
+	options[constant.IsPersistent] = false
+	options[constant.IsOfflinePush] = false
+	options[constant.IsUnreadCount] = false
+	options[constant.IsConversationUpdate] = false
+	options[constant.IsSenderSync] = false
+	options[constant.IsNotPrivate] = false
+	options[constant.IsSenderConversationUpdate] = false
+	options[constant.IsSenderNotificationPush] = false
+	options[constant.IsReactionFromCache] = false
+	for _, opt := range opts {
+		opt(options)
+	}
+	return options
+}
+
+func NewMsgOptions() Options {
+	options := make(map[string]bool, 11)
+	options[constant.IsOfflinePush] = false
+	return make(map[string]bool)
+}
+
+func WithOptions(options Options, opts ...OptionsOpt) Options {
+	for _, opt := range opts {
+		opt(options)
+	}
+	return options
+}
+
+func WithNotNotification(b bool) OptionsOpt {
+	return func(options Options) {
+		options[constant.IsNotNotification] = b
+	}
+}
+
+func WithSendMsg(b bool) OptionsOpt {
+	return func(options Options) {
+		options[constant.IsSendMsg] = b
+	}
+}
+
+func WithHistory(b bool) OptionsOpt {
+	return func(options Options) {
+		options[constant.IsHistory] = b
+	}
+}
+
+func WithPersistent() OptionsOpt {
+	return func(options Options) {
+		options[constant.IsPersistent] = true
+	}
+}
+
+func WithOfflinePush(b bool) OptionsOpt {
+	return func(options Options) {
+		options[constant.IsOfflinePush] = b
+	}
+}
+
+func WithUnreadCount(b bool) OptionsOpt {
+	return func(options Options) {
+		options[constant.IsUnreadCount] = b
+	}
+}
+
+func WithConversationUpdate() OptionsOpt {
+	return func(options Options) {
+		options[constant.IsConversationUpdate] = true
+	}
+}
+
+func WithSenderSync() OptionsOpt {
+	return func(options Options) {
+		options[constant.IsSenderSync] = true
+	}
+}
+
+func WithNotPrivate() OptionsOpt {
+	return func(options Options) {
+		options[constant.IsNotPrivate] = true
+	}
+}
+
+func WithSenderConversationUpdate() OptionsOpt {
+	return func(options Options) {
+		options[constant.IsSenderConversationUpdate] = true
+	}
+}
+
+func WithSenderNotificationPush() OptionsOpt {
+	return func(options Options) {
+		options[constant.IsSenderNotificationPush] = true
+	}
+}
+
+func WithReactionFromCache() OptionsOpt {
+	return func(options Options) {
+		options[constant.IsReactionFromCache] = true
+	}
+}
+
+func (o Options) Is(notification string) bool {
+	v, ok := o[notification]
+	if !ok || v {
+		return true
+	}
+	return false
+}
+
+func (o Options) IsNotNotification() bool {
+	return o.Is(constant.IsNotNotification)
+}
+
+func (o Options) IsSendMsg() bool {
+	return o.Is(constant.IsSendMsg)
+}
+
+func (o Options) IsHistory() bool {
+	return o.Is(constant.IsHistory)
+}
+
+func (o Options) IsPersistent() bool {
+	return o.Is(constant.IsPersistent)
+}
+
+func (o Options) IsOfflinePush() bool {
+	return o.Is(constant.IsOfflinePush)
+}
+
+func (o Options) IsUnreadCount() bool {
+	return o.Is(constant.IsUnreadCount)
+}
+
+func (o Options) IsConversationUpdate() bool {
+	return o.Is(constant.IsConversationUpdate)
+}
+
+func (o Options) IsSenderSync() bool {
+	return o.Is(constant.IsSenderSync)
+}
+
+func (o Options) IsNotPrivate() bool {
+	return o.Is(constant.IsNotPrivate)
+}
+
+func (o Options) IsSenderConversationUpdate() bool {
+	return o.Is(constant.IsSenderConversationUpdate)
+}
+
+func (o Options) IsSenderNotificationPush() bool {
+	return o.Is(constant.IsSenderNotificationPush)
+}
+
+func (o Options) IsReactionFromCache() bool {
+	return o.Is(constant.IsReactionFromCache)
+}