errcode

2026-05-21 01:09:01 +08:00 · 2023-02-15 15:52:32 +08:00
parent 85ebf24325
commit 6619183a48
62 changed files with 1319 additions and 2491 deletions
@@ -4,172 +4,97 @@ import (
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/db"
+	"Open_IM/pkg/common/db/cache"
+	"Open_IM/pkg/common/db/controller"
 	"Open_IM/pkg/common/db/mongo"
 	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/tracelog"
 	sdkws "Open_IM/pkg/proto/sdkws"
 	"Open_IM/pkg/utils"
+	"context"
 	"math"
 	"strconv"
 	"strings"

-	goRedis "github.com/go-redis/redis/v8"
+	"github.com/go-redis/redis/v8"
 	"github.com/golang/protobuf/proto"
 )

-const oldestList = 0
-const newestList = -1
+type SeqCheckInterface interface {
+	ClearAll() error
+}

-func ResetUserGroupMinSeq(operationID, groupID string, userIDList []string) error {
-	var delStruct delMsgRecursionStruct
-	minSeq, err := deleteMongoMsg(operationID, groupID, oldestList, &delStruct)
-	if err != nil {
-		log.NewError(operationID, utils.GetSelfFuncName(), groupID, "deleteMongoMsg failed")
+type ClearMsgCronTask struct {
+	msgModel   controller.MsgInterface
+	userModel  controller.UserInterface
+	groupModel controller.GroupInterface
+	cache      cache.Cache
+}
+
+func (c *ClearMsgCronTask) getCronTaskOperationID() string {
+	return cronTaskOperationID + utils.OperationIDGenerator()
+}
+
+func (c *ClearMsgCronTask) ClearAll() {
+	operationID := c.getCronTaskOperationID()
+	ctx := context.Background()
+	tracelog.SetOperationID(ctx, operationID)
+	log.NewInfo(operationID, "========================= start del cron task =========================")
+	var err error
+	userIDList, err := c.userModel.GetAllUserID(ctx)
+	if err == nil {
+		c.StartClearMsg(operationID, userIDList)
+	} else {
+		log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
 	}
-	if minSeq == 0 {
-		return nil
+	// working group msg clear
+	workingGroupIDList, err := im_mysql_model.GetGroupIDListByGroupType(constant.WorkingGroup)
+	if err == nil {
+		c.StartClearWorkingGroupMsg(operationID, workingGroupIDList)
+	} else {
+		log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
 	}
-	log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDList:", delStruct, "minSeq", minSeq)
+
+	log.NewInfo(operationID, "========================= start del cron finished =========================")
+}
+
+func (c *ClearMsgCronTask) StartClearMsg(operationID string, userIDList []string) {
+	log.NewDebug(operationID, utils.GetSelfFuncName(), "userIDList: ", userIDList)
 	for _, userID := range userIDList {
-		userMinSeq, err := db.DB.GetGroupUserMinSeq(groupID, userID)
-		if err != nil && err != goRedis.Nil {
-			log.NewError(operationID, utils.GetSelfFuncName(), "GetGroupUserMinSeq failed", groupID, userID, err.Error())
+		if err := DeleteUserMsgsAndSetMinSeq(operationID, userID); err != nil {
+			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), userID)
+		}
+		if err := checkMaxSeqWithMongo(operationID, userID, constant.WriteDiffusion); err != nil {
+			log.NewError(operationID, utils.GetSelfFuncName(), userID, err)
+		}
+	}
+}
+
+func (c *ClearMsgCronTask) StartClearWorkingGroupMsg(operationID string, workingGroupIDList []string) {
+	log.NewDebug(operationID, utils.GetSelfFuncName(), "workingGroupIDList: ", workingGroupIDList)
+	for _, groupID := range workingGroupIDList {
+		userIDList, err := rocksCache.GetGroupMemberIDListFromCache(groupID)
+		if err != nil {
+			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID)
 			continue
 		}
-		if userMinSeq > uint64(minSeq) {
-			err = db.DB.SetGroupUserMinSeq(groupID, userID, userMinSeq)
-		} else {
-			err = db.DB.SetGroupUserMinSeq(groupID, userID, uint64(minSeq))
+		log.NewDebug(operationID, utils.GetSelfFuncName(), "groupID:", groupID, "workingGroupIDList:", userIDList)
+		if err := DeleteUserSuperGroupMsgsAndSetMinSeq(operationID, groupID, userIDList); err != nil {
+			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userIDList)
 		}
-		if err != nil {
-			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userID, userMinSeq, minSeq)
+		if err := checkMaxSeqWithMongo(operationID, groupID, constant.ReadDiffusion); err != nil {
+			log.NewError(operationID, utils.GetSelfFuncName(), groupID, err)
 		}
 	}
-	return nil
 }

-func DeleteMongoMsgAndResetRedisSeq(operationID, userID string) error {
-	var delStruct delMsgRecursionStruct
-	minSeq, err := deleteMongoMsg(operationID, userID, oldestList, &delStruct)
-	if err != nil {
-		return utils.Wrap(err, "")
-	}
-	if minSeq == 0 {
-		return nil
-	}
-	log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDStruct: ", delStruct, "minSeq", minSeq)
-	err = db.DB.SetUserMinSeq(userID, minSeq)
-	return utils.Wrap(err, "")
-}
-
-// del list
-func delMongoMsgsPhysical(uidList []string) error {
-	if len(uidList) > 0 {
-		err := db.DB.DelMongoMsgs(uidList)
-		if err != nil {
-			return utils.Wrap(err, "DelMongoMsgs failed")
-		}
-	}
-	return nil
-}
-
-type delMsgRecursionStruct struct {
-	minSeq     uint32
-	delUidList []string
-}
-
-func (d *delMsgRecursionStruct) getSetMinSeq() uint32 {
-	return d.minSeq
-}
-
-// index 0....19(del) 20...69
-// seq 70
-// set minSeq 21
-// recursion 删除list并且返回设置的最小seq
-func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMsgRecursionStruct) (uint32, error) {
-	// find from oldest list
-	msgs, err := db.DB.GetUserMsgListByIndex(ID, index)
-	if err != nil || msgs.UID == "" {
-		if err != nil {
-			if err == mongoDB.ErrMsgListNotExist {
-				log.NewInfo(operationID, utils.GetSelfFuncName(), "ID:", ID, "index:", index, err.Error())
-			} else {
-				log.NewError(operationID, utils.GetSelfFuncName(), "GetUserMsgListByIndex failed", err.Error(), index, ID)
-			}
-		}
-		// 获取报错，或者获取不到了，物理删除并且返回seq
-		err = delMongoMsgsPhysical(delStruct.delUidList)
-		if err != nil {
-			return 0, err
-		}
-		return delStruct.getSetMinSeq() + 1, nil
-	}
-	log.NewDebug(operationID, "ID:", ID, "index:", index, "uid:", msgs.UID, "len:", len(msgs.Msg))
-	if len(msgs.Msg) > mongoDB.GetSingleGocMsgNum() {
-		log.NewWarn(operationID, utils.GetSelfFuncName(), "msgs too large", len(msgs.Msg), msgs.UID)
-	}
-	if msgs.Msg[len(msgs.Msg)-1].SendTime+(int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000) > utils.GetCurrentTimestampByMill() && msgListIsFull(msgs) {
-		delStruct.delUidList = append(delStruct.delUidList, msgs.UID)
-		lastMsgPb := &sdkws.MsgData{}
-		err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, lastMsgPb)
-		if err != nil {
-			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), len(msgs.Msg)-1, msgs.UID)
-			return 0, utils.Wrap(err, "proto.Unmarshal failed")
-		}
-		delStruct.minSeq = lastMsgPb.Seq
-	} else {
-		var hasMarkDelFlag bool
-		for _, msg := range msgs.Msg {
-			msgPb := &sdkws.MsgData{}
-			err = proto.Unmarshal(msg.Msg, msgPb)
-			if err != nil {
-				log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), len(msgs.Msg)-1, msgs.UID)
-				return 0, utils.Wrap(err, "proto.Unmarshal failed")
-			}
-			if utils.GetCurrentTimestampByMill() > msg.SendTime+(int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000) {
-				msgPb.Status = constant.MsgDeleted
-				bytes, _ := proto.Marshal(msgPb)
-				msg.Msg = bytes
-				msg.SendTime = 0
-				hasMarkDelFlag = true
-			} else {
-				if err := delMongoMsgsPhysical(delStruct.delUidList); err != nil {
-					return 0, err
-				}
-				if hasMarkDelFlag {
-					if err := db.DB.UpdateOneMsgList(msgs); err != nil {
-						return delStruct.getSetMinSeq(), utils.Wrap(err, "")
-					}
-				}
-				return msgPb.Seq + 1, nil
-			}
-		}
-	}
-	log.NewDebug(operationID, ID, "continue to", delStruct)
-	//  继续递归 index+1
-	seq, err := deleteMongoMsg(operationID, ID, index+1, delStruct)
-	return seq, utils.Wrap(err, "deleteMongoMsg failed")
-}
-
-func msgListIsFull(chat *mongoDB.UserChat) bool {
-	index, _ := strconv.Atoi(strings.Split(chat.UID, ":")[1])
-	if index == 0 {
-		if len(chat.Msg) >= 4999 {
-			return true
-		}
-	}
-	if len(chat.Msg) >= 5000 {
-		return true
-	}
-	return false
-}
-
-func checkMaxSeqWithMongo(operationID, ID string, diffusionType int) error {
+func checkMaxSeqWithMongo(operationID, sourceID string, diffusionType int) error {
 	var seqRedis uint64
 	var err error
 	if diffusionType == constant.WriteDiffusion {
-		seqRedis, err = db.DB.GetUserMaxSeq(ID)
+		seqRedis, err = db.DB.GetUserMaxSeq(sourceID)
 	} else {
-		seqRedis, err = db.DB.GetGroupMaxSeq(ID)
+		seqRedis, err = db.DB.GetGroupMaxSeq(sourceID)
 	}
 	if err != nil {
 		if err == goRedis.Nil {
@@ -177,7 +102,7 @@ func checkMaxSeqWithMongo(operationID, ID string, diffusionType int) error {
 		}
 		return utils.Wrap(err, "GetUserMaxSeq failed")
 	}
-	msg, err := db.DB.GetNewestMsg(ID)
+	msg, err := db.DB.GetNewestMsg(sourceID)
 	if err != nil {
 		return utils.Wrap(err, "GetNewestMsg failed")
 	}
@@ -185,9 +110,9 @@ func checkMaxSeqWithMongo(operationID, ID string, diffusionType int) error {
 		return nil
 	}
 	if math.Abs(float64(msg.Seq-uint32(seqRedis))) > 10 {
-		log.NewWarn(operationID, utils.GetSelfFuncName(), "seqMongo, seqRedis", msg.Seq, seqRedis, ID, "redis maxSeq is different with msg.Seq > 10", "status: ", msg.Status, msg.SendTime)
+		log.NewWarn(operationID, utils.GetSelfFuncName(), "seqMongo, seqRedis", msg.Seq, seqRedis, sourceID, "redis maxSeq is different with msg.Seq > 10", "status: ", msg.Status, msg.SendTime)
 	} else {
-		log.NewInfo(operationID, utils.GetSelfFuncName(), "seqMongo, seqRedis", msg.Seq, seqRedis, ID, "seq and msg OK", "status:", msg.Status, msg.SendTime)
+		log.NewInfo(operationID, utils.GetSelfFuncName(), "seqMongo, seqRedis", msg.Seq, seqRedis, sourceID, "seq and msg OK", "status:", msg.Status, msg.SendTime)
 	}
 	return nil
 }
@@ -2,8 +2,7 @@ package cronTask

 import (
 	"Open_IM/pkg/common/constant"
-	mongo2 "Open_IM/pkg/common/db/mongo"
-	sdkws "Open_IM/pkg/proto/sdkws"
+	"Open_IM/pkg/proto/sdkws"
 	"context"
 	"fmt"
 	"strconv"
@@ -59,7 +58,7 @@ func CreateChat(userChat *mongo2.UserChat) error {
 	return err
 }

-func TestDeleteMongoMsgAndResetRedisSeq(t *testing.T) {
+func TestDeleteUserMsgsAndSetMinSeq(t *testing.T) {
 	operationID := getCronTaskOperationID()
 	redisClient = redis.NewClient(&redis.Options{
 		Addr:     "127.0.0.1:16379",
@@ -81,7 +80,7 @@ func TestDeleteMongoMsgAndResetRedisSeq(t *testing.T) {
 	userChat := GenUserChat(1, 500, 200, 0, testUID1)
 	err = CreateChat(userChat)

-	if err := DeleteMongoMsgAndResetRedisSeq(operationID, testUID1); err != nil {
+	if err := DeleteUserMsgsAndSetMinSeq(operationID, testUID1); err != nil {
 		t.Error("checkMaxSeqWithMongo failed", testUID1)
 	}
 	if err := checkMaxSeqWithMongo(operationID, testUID1, constant.WriteDiffusion); err != nil {
@@ -94,7 +93,7 @@ func TestDeleteMongoMsgAndResetRedisSeq(t *testing.T) {
 	// for _, groupID := range testWorkingGroupIDList {
 	// 	operationID = groupID + "-" + operationID
 	// 	log.NewDebug(operationID, utils.GetSelfFuncName(), "groupID:", groupID, "userIDList:", testUserIDList)
-	// 	if err := ResetUserGroupMinSeq(operationID, groupID, testUserIDList); err != nil {
+	// 	if err := DeleteUserSuperGroupMsgsAndSetMinSeq(operationID, groupID, testUserIDList); err != nil {
 	// 		t.Error("checkMaxSeqWithMongo failed", groupID)
 	// 	}
 	// 	if err := checkMaxSeqWithMongo(operationID, groupID, constant.ReadDiffusion); err != nil {
@@ -2,10 +2,6 @@ package cronTask

 import (
 	"Open_IM/pkg/common/config"
-	"Open_IM/pkg/common/constant"
-	"Open_IM/pkg/common/db/controller"
-	"Open_IM/pkg/common/db/mysql_model/im_mysql_model"
-	rocksCache "Open_IM/pkg/common/db/rocks_cache"
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/utils"
 	"fmt"
@@ -15,9 +11,10 @@ import (
 )

 const cronTaskOperationID = "cronTaskOperationID-"
+const moduleName = "cron"

 func StartCronTask(userID, workingGroupID string) {
-	log.NewPrivateLog("cron")
+	log.NewPrivateLog(moduleName)
 	log.NewInfo(utils.OperationIDGenerator(), "start cron task", "cron config", config.Config.Mongo.ChatRecordsClearTime)
 	fmt.Println("cron task start, config", config.Config.Mongo.ChatRecordsClearTime)
 	if userID != "" {
@@ -44,59 +41,3 @@ func StartCronTask(userID, workingGroupID string) {
 		time.Sleep(10 * time.Second)
 	}
 }
-
-func getCronTaskOperationID() string {
-	return cronTaskOperationID + utils.OperationIDGenerator()
-}
-
-func ClearAll() {
-	operationID := getCronTaskOperationID()
-	log.NewInfo(operationID, "========================= start del cron task =========================")
-	var err error
-	userIDList, err := im_mysql_model.SelectAllUserID()
-	if err == nil {
-		StartClearMsg(operationID, userIDList)
-	} else {
-		log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
-	}
-
-	// working group msg clear
-	workingGroupIDList, err := im_mysql_model.GetGroupIDListByGroupType(constant.WorkingGroup)
-	if err == nil {
-		StartClearWorkingGroupMsg(operationID, workingGroupIDList)
-	} else {
-		log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
-	}
-
-	log.NewInfo(operationID, "========================= start del cron finished =========================")
-}
-
-func StartClearMsg(operationID string, userIDList []string) {
-	log.NewDebug(operationID, utils.GetSelfFuncName(), "userIDList: ", userIDList)
-	for _, userID := range userIDList {
-		if err := DeleteMongoMsgAndResetRedisSeq(operationID, userID); err != nil {
-			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), userID)
-		}
-		if err := checkMaxSeqWithMongo(operationID, userID, constant.WriteDiffusion); err != nil {
-			log.NewError(operationID, utils.GetSelfFuncName(), userID, err)
-		}
-	}
-}
-
-func StartClearWorkingGroupMsg(operationID string, workingGroupIDList []string) {
-	log.NewDebug(operationID, utils.GetSelfFuncName(), "workingGroupIDList: ", workingGroupIDList)
-	for _, groupID := range workingGroupIDList {
-		userIDList, err := rocksCache.GetGroupMemberIDListFromCache(groupID)
-		if err != nil {
-			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID)
-			continue
-		}
-		log.NewDebug(operationID, utils.GetSelfFuncName(), "groupID:", groupID, "workingGroupIDList:", userIDList)
-		if err := ResetUserGroupMinSeq(operationID, groupID, userIDList); err != nil {
-			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userIDList)
-		}
-		if err := checkMaxSeqWithMongo(operationID, groupID, constant.ReadDiffusion); err != nil {
-			log.NewError(operationID, utils.GetSelfFuncName(), groupID, err)
-		}
-	}
-}
@@ -8,7 +8,7 @@ import (
 	"fmt"
 	"sync"

-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"

 	"github.com/go-playground/validator/v10"
 )
@@ -40,7 +40,7 @@ func Run(promethuesPort int) {
 	go ws.run()
 	go rpcSvr.run()
 	go func() {
-		err := promePkg.StartPromeSrv(promethuesPort)
+		err := prome.StartPromeSrv(promethuesPort)
 		if err != nil {
 			panic(err)
 		}
@@ -5,7 +5,7 @@ import (
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/db"
 	"Open_IM/pkg/common/log"
-	promePkg "Open_IM/pkg/common/prometheus"
+	"Open_IM/pkg/common/prome"
 	pbChat "Open_IM/pkg/proto/msg"
 	push "Open_IM/pkg/proto/push"
 	pbRtc "Open_IM/pkg/proto/rtc"
@@ -51,18 +51,18 @@ func (ws *WServer) msgParse(conn *UserConn, binaryMsg []byte) {
 	case constant.WSGetNewestSeq:
 		log.NewInfo(m.OperationID, "getSeqReq ", m.SendID, m.MsgIncr, m.ReqIdentifier)
 		ws.getSeqReq(conn, &m)
-		promePkg.PromeInc(promePkg.GetNewestSeqTotalCounter)
+		prome.PromeInc(prome.GetNewestSeqTotalCounter)
 	case constant.WSSendMsg:
 		log.NewInfo(m.OperationID, "sendMsgReq ", m.SendID, m.MsgIncr, m.ReqIdentifier)
 		ws.sendMsgReq(conn, &m)
-		promePkg.PromeInc(promePkg.MsgRecvTotalCounter)
+		prome.PromeInc(prome.MsgRecvTotalCounter)
 	case constant.WSSendSignalMsg:
 		log.NewInfo(m.OperationID, "sendSignalMsgReq ", m.SendID, m.MsgIncr, m.ReqIdentifier)
 		ws.sendSignalMsgReq(conn, &m)
 	case constant.WSPullMsgBySeqList:
 		log.NewInfo(m.OperationID, "pullMsgBySeqListReq ", m.SendID, m.MsgIncr, m.ReqIdentifier)
 		ws.pullMsgBySeqListReq(conn, &m)
-		promePkg.PromeInc(promePkg.PullMsgBySeqListTotalCounter)
+		prome.PromeInc(prome.PullMsgBySeqListTotalCounter)
 	case constant.WsLogoutMsg:
 		log.NewInfo(m.OperationID, "conn.Close()", m.SendID, m.MsgIncr, m.ReqIdentifier)
 		ws.userLogoutReq(conn, &m)
@@ -4,7 +4,7 @@ import (
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/log"
-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prome"
 	"Open_IM/pkg/common/tokenverify"
 	pbRelay "Open_IM/pkg/proto/relay"
 	sdkws "Open_IM/pkg/proto/sdkws"
@@ -34,14 +34,14 @@ type RPCServer struct {
 }

 func initPrometheus() {
-	promePkg.NewMsgRecvTotalCounter()
-	promePkg.NewGetNewestSeqTotalCounter()
-	promePkg.NewPullMsgBySeqListTotalCounter()
-	promePkg.NewMsgOnlinePushSuccessCounter()
-	promePkg.NewOnlineUserGauges()
-	//promePkg.NewSingleChatMsgRecvSuccessCounter()
-	//promePkg.NewGroupChatMsgRecvSuccessCounter()
-	//promePkg.NewWorkSuperGroupChatMsgRecvSuccessCounter()
+	prome.NewMsgRecvTotalCounter()
+	prome.NewGetNewestSeqTotalCounter()
+	prome.NewPullMsgBySeqListTotalCounter()
+	prome.NewMsgOnlinePushSuccessCounter()
+	prome.NewOnlineUserGauges()
+	//prome.NewSingleChatMsgRecvSuccessCounter()
+	//prome.NewGroupChatMsgRecvSuccessCounter()
+	//prome.NewWorkSuperGroupChatMsgRecvSuccessCounter()
 }

 func (r *RPCServer) onInit(rpcPort int) {
@@ -67,11 +67,11 @@ func (r *RPCServer) run() {
 	defer listener.Close()
 	var grpcOpts []grpc.ServerOption
 	if config.Config.Prometheus.Enable {
-		promePkg.NewGrpcRequestCounter()
-		promePkg.NewGrpcRequestFailedCounter()
-		promePkg.NewGrpcRequestSuccessCounter()
+		prome.NewGrpcRequestCounter()
+		prome.NewGrpcRequestFailedCounter()
+		prome.NewGrpcRequestSuccessCounter()
 		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
+			// grpc.UnaryInterceptor(prome.UnaryServerInterceptorProme),
 			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
 			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
 		}...)
@@ -205,7 +205,7 @@ func (r *RPCServer) SuperGroupOnlineBatchPushOneMsg(_ context.Context, req *pbRe
 					resultCode := sendMsgBatchToUser(userConn, replyBytes.Bytes(), req, platform, v)
 					if resultCode == 0 && utils.IsContainInt(platform, r.pushTerminal) {
 						tempT.OnlinePush = true
-						promePkg.PromeInc(promePkg.MsgOnlinePushSuccessCounter)
+						prome.PromeInc(prome.MsgOnlinePushSuccessCounter)
 						log.Info(req.OperationID, "PushSuperMsgToUser is success By Ws", "args", req.String(), "recvPlatForm", constant.PlatformIDToName(platform), "recvID", v)
 						temp.ResultCode = resultCode
 						resp = append(resp, temp)
@@ -5,7 +5,7 @@ import (
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/db"
 	"Open_IM/pkg/common/log"
-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/common/tokenverify"
 	pbRelay "Open_IM/pkg/proto/relay"
 	"Open_IM/pkg/utils"
@@ -352,7 +352,7 @@ func (ws *WServer) addUserConn(uid string, platformID int, conn *UserConn, token
 	for _, v := range ws.wsUserToConn {
 		count = count + len(v)
 	}
-	promePkg.PromeGaugeInc(promePkg.OnlineUserGauge)
+	prome.PromeGaugeInc(prome.OnlineUserGauge)
 	log.Debug(operationID, "WS Add operation", "", "wsUser added", ws.wsUserToConn, "connection_uid", uid, "connection_platform", constant.PlatformIDToName(platformID), "online_user_num", len(ws.wsUserToConn), "online_conn_num", count)
 }

@@ -394,7 +394,7 @@ func (ws *WServer) delUserConn(conn *UserConn) {
 	if callbackResp.ErrCode != 0 {
 		log.NewError(operationID, utils.GetSelfFuncName(), "callbackUserOffline failed", callbackResp)
 	}
-	promePkg.PromeGaugeDec(promePkg.OnlineUserGauge)
+	prome.PromeGaugeDec(prome.OnlineUserGauge)

 }

@@ -1,29 +0,0 @@
-package msgtransfer
-
-import (
-	"Open_IM/pkg/common/db"
-	"Open_IM/pkg/common/log"
-	pbMsg "Open_IM/pkg/proto/msg"
-	"Open_IM/pkg/utils"
-)
-
-func saveUserChat(uid string, msg *pbMsg.MsgDataToMQ) error {
-	time := utils.GetCurrentTimestampByMill()
-	seq, err := db.DB.IncrUserSeq(uid)
-	if err != nil {
-		log.NewError(msg.OperationID, "data insert to redis err", err.Error(), msg.String())
-		return err
-	}
-	msg.MsgData.Seq = uint32(seq)
-	pbSaveData := pbMsg.MsgDataToDB{}
-	pbSaveData.MsgData = msg.MsgData
-	log.NewInfo(msg.OperationID, "IncrUserSeq cost time", utils.GetCurrentTimestampByMill()-time)
-	return db.DB.SaveUserChatMongo2(uid, pbSaveData.MsgData.SendTime, &pbSaveData)
-	//	return db.DB.SaveUserChatMongo2(uid, pbSaveData.MsgData.SendTime, &pbSaveData)
-}
-
-func saveUserChatList(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) (error, uint64) {
-	log.Info(operationID, utils.GetSelfFuncName(), "args ", userID, len(msgList))
-	//return db.DB.BatchInsertChat(userID, msgList, operationID)
-	return db.DB.BatchInsertChat2Cache(userID, msgList, operationID)
-}
@@ -2,83 +2,53 @@ package msgtransfer

 import (
 	"Open_IM/pkg/common/config"
-	"Open_IM/pkg/common/constant"
-	"Open_IM/pkg/common/kafka"
-	promePkg "Open_IM/pkg/common/prometheus"
-	"Open_IM/pkg/statistics"
+	"Open_IM/pkg/common/prome"
 	"fmt"
-	"sync"
 )

-const OnlineTopicBusy = 1
-const OnlineTopicVacancy = 0
-const Msg = 2
-const ConsumerMsgs = 3
-const AggregationMessages = 4
-const MongoMessages = 5
-const ChannelNum = 100
-
-var (
-	persistentCH          PersistentConsumerHandler
-	historyCH             OnlineHistoryRedisConsumerHandler
-	historyMongoCH        OnlineHistoryMongoConsumerHandler
-	modifyCH              ModifyMsgConsumerHandler
-	producer              *kafka.Producer
-	producerToModify      *kafka.Producer
-	producerToMongo       *kafka.Producer
-	cmdCh                 chan Cmd2Value
-	onlineTopicStatus     int
-	w                     *sync.Mutex
-	singleMsgSuccessCount uint64
-	groupMsgCount         uint64
-	singleMsgFailedCount  uint64
-
-	singleMsgSuccessCountMutex sync.Mutex
-)
-
-func Init() {
-	cmdCh = make(chan Cmd2Value, 10000)
-	w = new(sync.Mutex)
-	if config.Config.Prometheus.Enable {
-		initPrometheus()
-	}
-	persistentCH.Init()   // ws2mschat save mysql
-	historyCH.Init(cmdCh) //
-	historyMongoCH.Init()
-	modifyCH.Init()
-	onlineTopicStatus = OnlineTopicVacancy
-	//offlineHistoryCH.Init(cmdCh)
-	statistics.NewStatistics(&singleMsgSuccessCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second singleMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
-	statistics.NewStatistics(&groupMsgCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second groupMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
-	producer = kafka.NewKafkaProducer(config.Config.Kafka.Ms2pschat.Addr, config.Config.Kafka.Ms2pschat.Topic)
-	producerToModify = kafka.NewKafkaProducer(config.Config.Kafka.MsgToModify.Addr, config.Config.Kafka.MsgToModify.Topic)
-	producerToMongo = kafka.NewKafkaProducer(config.Config.Kafka.MsgToMongo.Addr, config.Config.Kafka.MsgToMongo.Topic)
+type MsgTransfer struct {
+	persistentCH   PersistentConsumerHandler         // 聊天记录持久化到mysql的消费者 订阅的topic: ws2ms_chat
+	historyCH      OnlineHistoryRedisConsumerHandler // 这个消费者聚合消息, 订阅的topic：ws2ms_chat, 修改通知发往msg_to_modify topic, 消息存入redis后Incr Redis, 再发消息到ms2pschat topic推送， 发消息到msg_to_mongo topic持久化
+	historyMongoCH OnlineHistoryMongoConsumerHandler // mongoDB批量插入, 成功后删除redis中消息，以及处理删除通知消息删除的 订阅的topic: msg_to_mongo
+	modifyCH       ModifyMsgConsumerHandler          // 负责消费修改消息通知的consumer, 订阅的topic: msg_to_modify
 }
-func Run(promethuesPort int) {
-	//register mysqlConsumerHandler to
-	if config.Config.ChatPersistenceMysql {
-		go persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(&persistentCH)
-	} else {
-		fmt.Println("not start mysql consumer")
+
+func NewMsgTransfer() *MsgTransfer {
+	msgTransfer := &MsgTransfer{}
+	msgTransfer.persistentCH.Init()
+	msgTransfer.historyCH.Init()
+	msgTransfer.historyMongoCH.Init()
+	msgTransfer.modifyCH.Init()
+	if config.Config.Prometheus.Enable {
+		msgTransfer.initPrometheus()
 	}
-	go historyCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyCH)
-	go historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyMongoCH)
-	go modifyCH.modifyMsgConsumerGroup.RegisterHandleAndConsumer(&modifyCH)
-	//go offlineHistoryCH.historyConsumerGroup.RegisterHandleAndConsumer(&offlineHistoryCH)
+	return msgTransfer
+}
+
+func (m *MsgTransfer) initPrometheus() {
+	prome.NewSeqGetSuccessCounter()
+	prome.NewSeqGetFailedCounter()
+	prome.NewSeqSetSuccessCounter()
+	prome.NewSeqSetFailedCounter()
+	prome.NewMsgInsertRedisSuccessCounter()
+	prome.NewMsgInsertRedisFailedCounter()
+	prome.NewMsgInsertMongoSuccessCounter()
+	prome.NewMsgInsertMongoFailedCounter()
+}
+
+func (m *MsgTransfer) Run(promePort int) {
+	if config.Config.ChatPersistenceMysql {
+		go m.persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(&m.persistentCH)
+	} else {
+		fmt.Println("msg transfer not start mysql consumer")
+	}
+	go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(&m.historyCH)
+	go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(&m.historyMongoCH)
+	go m.modifyCH.modifyMsgConsumerGroup.RegisterHandleAndConsumer(&m.modifyCH)
 	go func() {
-		err := promePkg.StartPromeSrv(promethuesPort)
+		err := prome.StartPromeSrv(promePort)
 		if err != nil {
 			panic(err)
 		}
 	}()
 }
-func SetOnlineTopicStatus(status int) {
-	w.Lock()
-	defer w.Unlock()
-	onlineTopicStatus = status
-}
-func GetOnlineTopicStatus() int {
-	w.Lock()
-	defer w.Unlock()
-	return onlineTopicStatus
-}
@@ -1,14 +1,19 @@
 package msgtransfer

 import (
+	"Open_IM/pkg/apistruct"
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
-	"Open_IM/pkg/common/db"
+	"Open_IM/pkg/common/db/cache"
+	"Open_IM/pkg/common/db/controller"
+	unRelationTb "Open_IM/pkg/common/db/table/unrelation"
 	kfk "Open_IM/pkg/common/kafka"
 	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/tracelog"
 	pbMsg "Open_IM/pkg/proto/msg"
 	sdkws "Open_IM/pkg/proto/sdkws"
 	"Open_IM/pkg/utils"
+	"context"
 	"encoding/json"
 	"github.com/Shopify/sarama"

@@ -16,13 +21,13 @@ import (
 )

 type ModifyMsgConsumerHandler struct {
-	msgHandle              map[string]fcb
 	modifyMsgConsumerGroup *kfk.MConsumerGroup
+
+	extendMsgInterface controller.ExtendMsgInterface
+	cache              cache.Cache
 }

 func (mmc *ModifyMsgConsumerHandler) Init() {
-	mmc.msgHandle = make(map[string]fcb)
-	mmc.msgHandle[config.Config.Kafka.MsgToModify.Topic] = mmc.ModifyMsg
 	mmc.modifyMsgConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
 		OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.MsgToModify.Topic},
 		config.Config.Kafka.MsgToModify.Addr, config.Config.Kafka.ConsumerGroupID.MsgToModify)
@@ -35,7 +40,7 @@ func (mmc *ModifyMsgConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSessi
 	for msg := range claim.Messages() {
 		log.NewDebug("", "kafka get info to mysql", "ModifyMsgConsumerHandler", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
 		if len(msg.Value) != 0 {
-			mmc.msgHandle[msg.Topic](msg, string(msg.Key), sess)
+			mmc.ModifyMsg(msg, string(msg.Key), sess)
 		} else {
 			log.Error("", "msg get from kafka but is nil", msg.Key)
 		}
@@ -58,6 +63,8 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(cMsg *sarama.ConsumerMessage, msg
 		if !isReactionFromCache {
 			continue
 		}
+		ctx := context.Background()
+		tracelog.SetOperationID(ctx, msgDataToMQ.OperationID)
 		if msgDataToMQ.MsgData.ContentType == constant.ReactionMessageModifier {
 			notification := &apistruct.ReactionMessageModifierNotification{}
 			if err := json.Unmarshal(msgDataToMQ.MsgData.Content, notification); err != nil {
@@ -69,21 +76,21 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(cMsg *sarama.ConsumerMessage, msg
 			}
 			if !notification.IsReact {
 				// first time to modify
-				var reactionExtensionList = make(map[string]mongoDB.KeyValue)
-				extendMsg := mongoDB.ExtendMsg{
+				var reactionExtensionList = make(map[string]unRelationTb.KeyValueModel)
+				extendMsg := unRelationTb.ExtendMsgModel{
 					ReactionExtensionList: reactionExtensionList,
 					ClientMsgID:           notification.ClientMsgID,
 					MsgFirstModifyTime:    notification.MsgFirstModifyTime,
 				}
 				for _, v := range notification.SuccessReactionExtensionList {
-					reactionExtensionList[v.TypeKey] = mongoDB.KeyValue{
+					reactionExtensionList[v.TypeKey] = unRelationTb.KeyValueModel{
 						TypeKey:          v.TypeKey,
 						Value:            v.Value,
 						LatestUpdateTime: v.LatestUpdateTime,
 					}
 				}

-				if err := db.DB.InsertExtendMsg(notification.SourceID, notification.SessionType, &extendMsg); err != nil {
+				if err := mmc.extendMsgInterface.InsertExtendMsg(ctx, notification.SourceID, notification.SessionType, &extendMsg); err != nil {
 					log.NewError(msgDataToMQ.OperationID, "MsgFirstModify InsertExtendMsg failed", notification.SourceID, notification.SessionType, extendMsg, err.Error())
 					continue
 				}
@@ -97,7 +104,7 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(cMsg *sarama.ConsumerMessage, msg
 					}
 				}
 				// is already modify
-				if err := db.DB.InsertOrUpdateReactionExtendMsgSet(notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, reactionExtensionList); err != nil {
+				if err := mmc.extendMsgInterface.InsertOrUpdateReactionExtendMsgSet(ctx, notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, reactionExtensionList); err != nil {
 					log.NewError(msgDataToMQ.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
 				}
 			}
@@ -106,15 +113,10 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(cMsg *sarama.ConsumerMessage, msg
 			if err := json.Unmarshal(msgDataToMQ.MsgData.Content, notification); err != nil {
 				continue
 			}
-			if err := db.DB.DeleteReactionExtendMsgSet(notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, notification.SuccessReactionExtensionList); err != nil {
+			if err := mmc.extendMsgInterface.DeleteReactionExtendMsgSet(ctx, notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, notification.SuccessReactionExtensionList); err != nil {
 				log.NewError(msgDataToMQ.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
 			}
 		}
 	}

 }
-
-func UnMarshallSetReactionMsgContent(content []byte) (notification *apistruct.ReactionMessageModifierNotification, err error) {
-
-	return notification, nil
-}
@@ -3,62 +3,80 @@ package msgtransfer
 import (
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
-	"Open_IM/pkg/common/db"
-	kfk "Open_IM/pkg/common/kafka"
+	"Open_IM/pkg/common/db/cache"
+	"Open_IM/pkg/common/db/controller"
+	"Open_IM/pkg/common/kafka"
 	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/tracelog"
 	pbMsg "Open_IM/pkg/proto/msg"
 	pbPush "Open_IM/pkg/proto/push"
+	"Open_IM/pkg/statistics"
 	"Open_IM/pkg/utils"
 	"context"
+	"fmt"
 	"github.com/Shopify/sarama"
 	"github.com/golang/protobuf/proto"
-	"hash/crc32"
-	"strings"
 	"sync"
 	"time"
 )

+const ConsumerMsgs = 3
+const AggregationMessages = 4
+const MongoMessages = 5
+const ChannelNum = 100
+
 type MsgChannelValue struct {
 	aggregationID string //maybe userID or super groupID
 	triggerID     string
 	msgList       []*pbMsg.MsgDataToMQ
 	lastSeq       uint64
 }
+
 type TriggerChannelValue struct {
 	triggerID string
-	cmsgList  []*sarama.ConsumerMessage
+	cMsgList  []*sarama.ConsumerMessage
 }
-type fcb func(cMsg *sarama.ConsumerMessage, msgKey string, sess sarama.ConsumerGroupSession)
+
 type Cmd2Value struct {
 	Cmd   int
 	Value interface{}
 }
+
 type OnlineHistoryRedisConsumerHandler struct {
-	msgHandle            map[string]fcb
-	historyConsumerGroup *kfk.MConsumerGroup
+	historyConsumerGroup *kafka.MConsumerGroup
 	chArrays             [ChannelNum]chan Cmd2Value
 	msgDistributionCh    chan Cmd2Value
+
+	singleMsgSuccessCount      uint64
+	singleMsgFailedCount       uint64
+	singleMsgSuccessCountMutex sync.Mutex
+	singleMsgFailedCountMutex  sync.Mutex
+
+	producerToPush   *kafka.Producer
+	producerToModify *kafka.Producer
+	producerToMongo  *kafka.Producer
+
+	msgInterface controller.MsgInterface
+	cache        cache.Cache
 }

-func (och *OnlineHistoryRedisConsumerHandler) Init(cmdCh chan Cmd2Value) {
-	och.msgHandle = make(map[string]fcb)
+func (och *OnlineHistoryRedisConsumerHandler) Init() {
 	och.msgDistributionCh = make(chan Cmd2Value) //no buffer channel
 	go och.MessagesDistributionHandle()
 	for i := 0; i < ChannelNum; i++ {
 		och.chArrays[i] = make(chan Cmd2Value, 50)
 		go och.Run(i)
 	}
-	if config.Config.ReliableStorage {
-		och.msgHandle[config.Config.Kafka.Ws2mschat.Topic] = och.handleChatWs2Mongo
-	} else {
-		och.msgHandle[config.Config.Kafka.Ws2mschat.Topic] = och.handleChatWs2MongoLowReliability
-
-	}
-	och.historyConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
+	och.producerToPush = kafka.NewKafkaProducer(config.Config.Kafka.Ms2pschat.Addr, config.Config.Kafka.Ms2pschat.Topic)
+	och.producerToModify = kafka.NewKafkaProducer(config.Config.Kafka.MsgToModify.Addr, config.Config.Kafka.MsgToModify.Topic)
+	och.producerToMongo = kafka.NewKafkaProducer(config.Config.Kafka.MsgToMongo.Addr, config.Config.Kafka.MsgToMongo.Topic)
+	och.historyConsumerGroup = kafka.NewMConsumerGroup(&kafka.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
 		OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ws2mschat.Topic},
 		config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToRedis)

+	statistics.NewStatistics(&och.singleMsgSuccessCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second singleMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
 }
+
 func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 	for {
 		select {
@@ -72,6 +90,8 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 				notStoragePushMsgList := make([]*pbMsg.MsgDataToMQ, 0, 80)
 				log.Debug(triggerID, "msg arrived channel", "channel id", channelID, msgList, msgChannelValue.aggregationID, len(msgList))
 				var modifyMsgList []*pbMsg.MsgDataToMQ
+				ctx := context.Background()
+				tracelog.SetOperationID(ctx, triggerID)
 				for _, v := range msgList {
 					log.Debug(triggerID, "msg come to storage center", v.String())
 					isHistory := utils.GetSwitchFromOptions(v.MsgData.Options, constant.IsHistory)
@@ -84,45 +104,36 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 							notStoragePushMsgList = append(notStoragePushMsgList, v)
 						}
 					}
-
 					if v.MsgData.ContentType == constant.ReactionMessageModifier || v.MsgData.ContentType == constant.ReactionMessageDeleter {
 						modifyMsgList = append(modifyMsgList, v)
 					}
 				}
 				if len(modifyMsgList) > 0 {
-					sendMessageToModifyMQ(msgChannelValue.aggregationID, triggerID, modifyMsgList)
+					och.sendMessageToModifyMQ(ctx, msgChannelValue.aggregationID, triggerID, modifyMsgList)
 				}
-				//switch msgChannelValue.msg.MsgData.SessionType {
-				//case constant.SingleChatType:
-				//case constant.GroupChatType:
-				//case constant.NotificationChatType:
-				//default:
-				//	log.NewError(msgFromMQ.OperationID, "SessionType error", msgFromMQ.String())
-				//	return
-				//}
 				log.Debug(triggerID, "msg storage length", len(storageMsgList), "push length", len(notStoragePushMsgList))
 				if len(storageMsgList) > 0 {
-					err, lastSeq := saveUserChatList(msgChannelValue.aggregationID, storageMsgList, triggerID)
+					lastSeq, err := och.msgInterface.BatchInsertChat2Cache(ctx, msgChannelValue.aggregationID, storageMsgList)
 					if err != nil {
-						singleMsgFailedCount += uint64(len(storageMsgList))
+						och.singleMsgFailedCountMutex.Lock()
+						och.singleMsgFailedCount += uint64(len(storageMsgList))
+						och.singleMsgFailedCountMutex.Unlock()
 						log.NewError(triggerID, "single data insert to redis err", err.Error(), storageMsgList)
 					} else {
-						singleMsgSuccessCountMutex.Lock()
-						singleMsgSuccessCount += uint64(len(storageMsgList))
-						singleMsgSuccessCountMutex.Unlock()
-						och.SendMessageToMongoCH(msgChannelValue.aggregationID, triggerID, storageMsgList, lastSeq)
-
+						och.singleMsgSuccessCountMutex.Lock()
+						och.singleMsgSuccessCount += uint64(len(storageMsgList))
+						och.singleMsgSuccessCountMutex.Unlock()
+						och.SendMessageToMongoCH(ctx, msgChannelValue.aggregationID, triggerID, storageMsgList, lastSeq)
 						for _, v := range storageMsgList {
-							sendMessageToPushMQ(v, msgChannelValue.aggregationID)
+							och.sendMessageToPushMQ(ctx, v, msgChannelValue.aggregationID)
 						}
 						for _, x := range notStoragePushMsgList {
-							sendMessageToPushMQ(x, msgChannelValue.aggregationID)
+							och.sendMessageToPushMQ(ctx, x, msgChannelValue.aggregationID)
 						}
 					}
-
 				} else {
-					for _, x := range notStoragePushMsgList {
-						sendMessageToPushMQ(x, msgChannelValue.aggregationID)
+					for _, v := range notStoragePushMsgList {
+						och.sendMessageToPushMQ(ctx, v, msgChannelValue.aggregationID)
 					}
 				}
 			}
@@ -130,62 +141,6 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 	}
 }

-func (och *OnlineHistoryRedisConsumerHandler) SendMessageToMongoCH(aggregationID string, triggerID string, messages []*pbMsg.MsgDataToMQ, lastSeq uint64) {
-	if len(messages) > 0 {
-		pid, offset, err := producerToMongo.SendMessage(&pbMsg.MsgDataToMongoByMQ{LastSeq: lastSeq, AggregationID: aggregationID, MessageList: messages, TriggerID: triggerID}, aggregationID, triggerID)
-		if err != nil {
-			log.Error(triggerID, "kafka send failed", "send data", len(messages), "pid", pid, "offset", offset, "err", err.Error(), "key", aggregationID)
-		} else {
-			//	log.NewWarn(m.OperationID, "sendMsgToKafka   client msgID ", m.MsgData.ClientMsgID)
-		}
-	}
-	//hashCode := getHashCode(aggregationID)
-	//channelID := hashCode % ChannelNum
-	//log.Debug(triggerID, "generate channelID", hashCode, channelID, aggregationID)
-	////go func(cID uint32, userID string, messages []*pbMsg.MsgDataToMQ) {
-	//och.chMongoArrays[channelID] <- Cmd2Value{Cmd: MongoMessages, Value: MsgChannelValue{aggregationID: aggregationID, msgList: messages, triggerID: triggerID, lastSeq: lastSeq}}
-}
-
-//func (och *OnlineHistoryRedisConsumerHandler) MongoMessageRun(channelID int) {
-//	for {
-//		select {
-//		case cmd := <-och.chMongoArrays[channelID]:
-//			switch cmd.Cmd {
-//			case MongoMessages:
-//				msgChannelValue := cmd.Value.(MsgChannelValue)
-//				msgList := msgChannelValue.msgList
-//				triggerID := msgChannelValue.triggerID
-//				aggregationID := msgChannelValue.aggregationID
-//				lastSeq := msgChannelValue.lastSeq
-//				err := db.DB.BatchInsertChat2DB(aggregationID, msgList, triggerID, lastSeq)
-//				if err != nil {
-//					log.NewError(triggerID, "single data insert to mongo err", err.Error(), msgList)
-//				}
-//				for _, v := range msgList {
-//					if v.MsgData.ContentType == constant.DeleteMessageNotification {
-//						tips := sdkws.TipsComm{}
-//						DeleteMessageTips := sdkws.DeleteMessageTips{}
-//						err := proto.Unmarshal(v.MsgData.Content, &tips)
-//						if err != nil {
-//							log.NewError(triggerID, "tips unmarshal err:", err.Error(), v.String())
-//							continue
-//						}
-//						err = proto.Unmarshal(tips.Detail, &DeleteMessageTips)
-//						if err != nil {
-//							log.NewError(triggerID, "deleteMessageTips unmarshal err:", err.Error(), v.String())
-//							continue
-//						}
-//						if unexistSeqList, err := db.DB.DelMsgBySeqList(DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID); err != nil {
-//							log.NewError(v.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqList args: ", DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID, err.Error(), unexistSeqList)
-//						}
-//
-//					}
-//				}
-//			}
-//		}
-//	}
-//}
-
 func (och *OnlineHistoryRedisConsumerHandler) MessagesDistributionHandle() {
 	for {
 		aggregationMsgs := make(map[string][]*pbMsg.MsgDataToMQ, ChannelNum)
@@ -195,7 +150,7 @@ func (och *OnlineHistoryRedisConsumerHandler) MessagesDistributionHandle() {
 			case ConsumerMsgs:
 				triggerChannelValue := cmd.Value.(TriggerChannelValue)
 				triggerID := triggerChannelValue.triggerID
-				consumerMessages := triggerChannelValue.cmsgList
+				consumerMessages := triggerChannelValue.cMsgList
 				//Aggregation map[userid]message list
 				log.Debug(triggerID, "batch messages come to distribution center", len(consumerMessages))
 				for i := 0; i < len(consumerMessages); i++ {
@@ -218,155 +173,21 @@ func (och *OnlineHistoryRedisConsumerHandler) MessagesDistributionHandle() {
 				log.Debug(triggerID, "generate map list users len", len(aggregationMsgs))
 				for aggregationID, v := range aggregationMsgs {
 					if len(v) >= 0 {
-						hashCode := getHashCode(aggregationID)
+						hashCode := utils.GetHashCode(aggregationID)
 						channelID := hashCode % ChannelNum
 						log.Debug(triggerID, "generate channelID", hashCode, channelID, aggregationID)
-						//go func(cID uint32, userID string, messages []*pbMsg.MsgDataToMQ) {
 						och.chArrays[channelID] <- Cmd2Value{Cmd: AggregationMessages, Value: MsgChannelValue{aggregationID: aggregationID, msgList: v, triggerID: triggerID}}
-						//}(channelID, userID, v)
 					}
 				}
 			}
 		}
-
-	}
-
-}
-func (mc *OnlineHistoryRedisConsumerHandler) handleChatWs2Mongo(cMsg *sarama.ConsumerMessage, msgKey string, sess sarama.ConsumerGroupSession) {
-	msg := cMsg.Value
-	now := time.Now()
-	msgFromMQ := pbMsg.MsgDataToMQ{}
-	err := proto.Unmarshal(msg, &msgFromMQ)
-	if err != nil {
-		log.Error("msg_transfer Unmarshal msg err", "", "msg", string(msg), "err", err.Error())
-		return
-	}
-	operationID := msgFromMQ.OperationID
-	log.NewInfo(operationID, "msg come mongo!!!", "", "msg", string(msg))
-	//Control whether to store offline messages (mongo)
-	isHistory := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsHistory)
-	//Control whether to store history messages (mysql)
-	isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent)
-	isSenderSync := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsSenderSync)
-	switch msgFromMQ.MsgData.SessionType {
-	case constant.SingleChatType:
-		log.NewDebug(msgFromMQ.OperationID, "msg_transfer msg type = SingleChatType", isHistory, isPersist)
-		if isHistory {
-			err := saveUserChat(msgKey, &msgFromMQ)
-			if err != nil {
-				singleMsgFailedCount++
-				log.NewError(operationID, "single data insert to mongo err", err.Error(), msgFromMQ.String())
-				return
-			}
-			singleMsgSuccessCountMutex.Lock()
-			singleMsgSuccessCount++
-			singleMsgSuccessCountMutex.Unlock()
-			log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", time.Since(now))
-		}
-		if !isSenderSync && msgKey == msgFromMQ.MsgData.SendID {
-		} else {
-			go sendMessageToPush(&msgFromMQ, msgKey)
-		}
-		log.NewDebug(operationID, "saveSingleMsg cost time ", time.Since(now))
-	case constant.GroupChatType:
-		log.NewDebug(msgFromMQ.OperationID, "msg_transfer msg type = GroupChatType", isHistory, isPersist)
-		if isHistory {
-			err := saveUserChat(msgFromMQ.MsgData.RecvID, &msgFromMQ)
-			if err != nil {
-				log.NewError(operationID, "group data insert to mongo err", msgFromMQ.String(), msgFromMQ.MsgData.RecvID, err.Error())
-				return
-			}
-			groupMsgCount++
-		}
-		go sendMessageToPush(&msgFromMQ, msgFromMQ.MsgData.RecvID)
-		log.NewDebug(operationID, "saveGroupMsg cost time ", time.Since(now))
-
-	case constant.NotificationChatType:
-		log.NewDebug(msgFromMQ.OperationID, "msg_transfer msg type = NotificationChatType", isHistory, isPersist)
-		if isHistory {
-			err := saveUserChat(msgKey, &msgFromMQ)
-			if err != nil {
-				log.NewError(operationID, "single data insert to mongo err", err.Error(), msgFromMQ.String())
-				return
-			}
-			log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", time.Since(now))
-		}
-		if !isSenderSync && msgKey == msgFromMQ.MsgData.SendID {
-		} else {
-			go sendMessageToPush(&msgFromMQ, msgKey)
-		}
-		log.NewDebug(operationID, "saveUserChat cost time ", time.Since(now))
-	default:
-		log.NewError(msgFromMQ.OperationID, "SessionType error", msgFromMQ.String())
-		return
-	}
-	sess.MarkMessage(cMsg, "")
-	log.NewDebug(msgFromMQ.OperationID, "msg_transfer handle topic data to database success...", msgFromMQ.String())
-}
-
-func (och *OnlineHistoryRedisConsumerHandler) handleChatWs2MongoLowReliability(cMsg *sarama.ConsumerMessage, msgKey string, sess sarama.ConsumerGroupSession) {
-	msg := cMsg.Value
-	msgFromMQ := pbMsg.MsgDataToMQ{}
-	err := proto.Unmarshal(msg, &msgFromMQ)
-	if err != nil {
-		log.Error("msg_transfer Unmarshal msg err", "", "msg", string(msg), "err", err.Error())
-		return
-	}
-	operationID := msgFromMQ.OperationID
-	log.NewInfo(operationID, "msg come mongo!!!", "", "msg", string(msg))
-	//Control whether to store offline messages (mongo)
-	isHistory := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsHistory)
-	isSenderSync := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsSenderSync)
-	if isHistory {
-		seq, err := db.DB.IncrUserSeq(msgKey)
-		if err != nil {
-			log.NewError(operationID, "data insert to redis err", err.Error(), string(msg))
-			return
-		}
-		sess.MarkMessage(cMsg, "")
-		msgFromMQ.MsgData.Seq = uint32(seq)
-		log.Debug(operationID, "send ch msg is ", msgFromMQ.String())
-		//och.msgCh <- Cmd2Value{Cmd: Msg, Value: MsgChannelValue{msgKey, msgFromMQ}}
-		//err := saveUserChat(msgKey, &msgFromMQ)
-		//if err != nil {
-		//	singleMsgFailedCount++
-		//	log.NewError(operationID, "single data insert to mongo err", err.Error(), msgFromMQ.String())
-		//	return
-		//}
-		//singleMsgSuccessCountMutex.Lock()
-		//singleMsgSuccessCount++
-		//singleMsgSuccessCountMutex.Unlock()
-		//log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", time.Since(now))
-	} else {
-		if !(!isSenderSync && msgKey == msgFromMQ.MsgData.SendID) {
-			go sendMessageToPush(&msgFromMQ, msgKey)
-		}
 	}
 }

 func (OnlineHistoryRedisConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
 func (OnlineHistoryRedisConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }

-//func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
-//	claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
-//	log.NewDebug("", "online new session msg come", claim.HighWaterMarkOffset(), claim.Topic(), claim.Partition())
-//	for msg := range claim.Messages() {
-//		SetOnlineTopicStatus(OnlineTopicBusy)
-//		//och.TriggerCmd(OnlineTopicBusy)
-//		log.NewDebug("", "online kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "online", msg.Offset, claim.HighWaterMarkOffset())
-//		och.msgHandle[msg.Topic](msg, string(msg.Key), sess)
-//		if claim.HighWaterMarkOffset()-msg.Offset <= 1 {
-//			log.Debug("", "online msg consume end", claim.HighWaterMarkOffset(), msg.Offset)
-//			SetOnlineTopicStatus(OnlineTopicVacancy)
-//			och.TriggerCmd(OnlineTopicVacancy)
-//		}
-//	}
-//	return nil
-//}
-
-func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
-	claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
-
+func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
 	for {
 		if sess == nil {
 			log.NewWarn("", " sess == nil, waiting ")
@@ -383,24 +204,6 @@ func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerG
 	go func() {
 		for {
 			select {
-			//case :
-			//	triggerID = utils.OperationIDGenerator()
-			//
-			//	log.NewDebug(triggerID, "claim.Messages ", msg)
-			//	cMsg = append(cMsg, msg)
-			//	if len(cMsg) >= 1000 {
-			//		ccMsg := make([]*sarama.ConsumerMessage, 0, 1000)
-			//		for _, v := range cMsg {
-			//			ccMsg = append(ccMsg, v)
-			//		}
-			//		log.Debug(triggerID, "length trigger msg consumer start", len(ccMsg))
-			//		och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
-			//			triggerID: triggerID, cmsgList: ccMsg}}
-			//		sess.MarkMessage(msg, "")
-			//		cMsg = make([]*sarama.ConsumerMessage, 0, 1000)
-			//		log.Debug(triggerID, "length trigger msg consumer end", len(cMsg))
-			//	}
-
 			case <-t.C:
 				if len(cMsg) > 0 {
 					rwLock.Lock()
@@ -416,163 +219,53 @@ func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerG
 					for i := 0; i < len(ccMsg)/split; i++ {
 						//log.Debug()
 						och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
-							triggerID: triggerID, cmsgList: ccMsg[i*split : (i+1)*split]}}
+							triggerID: triggerID, cMsgList: ccMsg[i*split : (i+1)*split]}}
 					}
 					if (len(ccMsg) % split) > 0 {
 						och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
-							triggerID: triggerID, cmsgList: ccMsg[split*(len(ccMsg)/split):]}}
+							triggerID: triggerID, cMsgList: ccMsg[split*(len(ccMsg)/split):]}}
 					}
-					//sess.MarkMessage(ccMsg[len(cMsg)-1], "")
-
 					log.Debug(triggerID, "timer trigger msg consumer end", len(cMsg))
 				}
 			}
 		}
-
 	}()
 	for msg := range claim.Messages() {
-		//msgFromMQ := pbMsg.MsgDataToMQ{}
-		//err := proto.Unmarshal(msg.Value, &msgFromMQ)
-		//if err != nil {
-		//	log.Error(triggerID, "msg_transfer Unmarshal msg err", "msg", string(msg.Value), "err", err.Error())
-		//}
-		//userID := string(msg.Key)
-		//hashCode := getHashCode(userID)
-		//channelID := hashCode % ChannelNum
-		//log.Debug(triggerID, "generate channelID", hashCode, channelID, userID)
-		////go func(cID uint32, userID string, messages []*pbMsg.MsgDataToMQ) {
-		//och.chArrays[channelID] <- Cmd2Value{Cmd: UserMessages, Value: MsgChannelValue{userID: userID, msgList: []*pbMsg.MsgDataToMQ{&msgFromMQ}, triggerID: msgFromMQ.OperationID}}
-		//sess.MarkMessage(msg, "")
 		rwLock.Lock()
 		if len(msg.Value) != 0 {
 			cMsg = append(cMsg, msg)
 		}
 		rwLock.Unlock()
 		sess.MarkMessage(msg, "")
-		//och.TriggerCmd(OnlineTopicBusy)
-
-		//log.NewDebug("", "online kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "online", msg.Offset, claim.HighWaterMarkOffset())
-
 	}
-
 	return nil
 }

-//func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
-//	claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
-//
-//	for {
-//		if sess == nil {
-//			log.NewWarn("", " sess == nil, waiting ")
-//			time.Sleep(100 * time.Millisecond)
-//		} else {
-//			break
-//		}
-//	}
-//
-//	log.NewDebug("", "online new session msg come", claim.HighWaterMarkOffset(), claim.Topic(), claim.Partition())
-//	cMsg := make([]*sarama.ConsumerMessage, 0, 1000)
-//	t := time.NewTicker(time.Duration(100) * time.Millisecond)
-//	var triggerID string
-//	for msg := range claim.Messages() {
-//		cMsg = append(cMsg, msg)
-//		//och.TriggerCmd(OnlineTopicBusy)
-//		select {
-//		//case :
-//		//	triggerID = utils.OperationIDGenerator()
-//		//
-//		//	log.NewDebug(triggerID, "claim.Messages ", msg)
-//		//	cMsg = append(cMsg, msg)
-//		//	if len(cMsg) >= 1000 {
-//		//		ccMsg := make([]*sarama.ConsumerMessage, 0, 1000)
-//		//		for _, v := range cMsg {
-//		//			ccMsg = append(ccMsg, v)
-//		//		}
-//		//		log.Debug(triggerID, "length trigger msg consumer start", len(ccMsg))
-//		//		och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
-//		//			triggerID: triggerID, cmsgList: ccMsg}}
-//		//		sess.MarkMessage(msg, "")
-//		//		cMsg = make([]*sarama.ConsumerMessage, 0, 1000)
-//		//		log.Debug(triggerID, "length trigger msg consumer end", len(cMsg))
-//		//	}
-//
-//		case <-t.C:
-//			if len(cMsg) > 0 {
-//				ccMsg := make([]*sarama.ConsumerMessage, 0, 1000)
-//				for _, v := range cMsg {
-//					ccMsg = append(ccMsg, v)
-//				}
-//				triggerID = utils.OperationIDGenerator()
-//				log.Debug(triggerID, "timer trigger msg consumer start", len(ccMsg))
-//				och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
-//					triggerID: triggerID, cmsgList: ccMsg}}
-//				sess.MarkMessage(cMsg[len(cMsg)-1], "")
-//				cMsg = make([]*sarama.ConsumerMessage, 0, 1000)
-//				log.Debug(triggerID, "timer trigger msg consumer end", len(cMsg))
-//			}
-//		default:
-//
-//		}
-//		//log.NewDebug("", "online kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "online", msg.Offset, claim.HighWaterMarkOffset())
-//
-//	}
-//	return nil
-//}
-
-func sendMessageToPush(message *pbMsg.MsgDataToMQ, pushToUserID string) {
-	log.Info(message.OperationID, "msg_transfer send message to push", "message", message.String())
-	rpcPushMsg := pbPush.PushMsgReq{OperationID: message.OperationID, MsgData: message.MsgData, PushToUserID: pushToUserID}
-	mqPushMsg := pbMsg.PushMsgDataToMQ{OperationID: message.OperationID, MsgData: message.MsgData, PushToUserID: pushToUserID}
-	grpcConn := rpc.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImPushName, message.OperationID)
-	if grpcConn != nil {
-		log.Error(rpcPushMsg.OperationID, "rpc dial failed", "push data", rpcPushMsg.String())
-		pid, offset, err := producer.SendMessage(&mqPushMsg, mqPushMsg.PushToUserID, rpcPushMsg.OperationID)
-		if err != nil {
-			log.Error(mqPushMsg.OperationID, "kafka send failed", "send data", message.String(), "pid", pid, "offset", offset, "err", err.Error())
-		}
-		return
-	}
-	msgClient := pbPush.NewPushMsgServiceClient(grpcConn)
-	_, err := msgClient.PushMsg(context.Background(), &rpcPushMsg)
-	if err != nil {
-		log.Error(rpcPushMsg.OperationID, "rpc send failed", rpcPushMsg.OperationID, "push data", rpcPushMsg.String(), "err", err.Error())
-		pid, offset, err := producer.SendMessage(&mqPushMsg, mqPushMsg.PushToUserID, rpcPushMsg.OperationID)
-		if err != nil {
-			log.Error(message.OperationID, "kafka send failed", mqPushMsg.OperationID, "send data", mqPushMsg.String(), "pid", pid, "offset", offset, "err", err.Error())
-		}
-	} else {
-		log.Info(message.OperationID, "rpc send success", rpcPushMsg.OperationID, "push data", rpcPushMsg.String())
-
-	}
-}
-
-func sendMessageToPushMQ(message *pbMsg.MsgDataToMQ, pushToUserID string) {
+func (och *OnlineHistoryRedisConsumerHandler) sendMessageToPushMQ(ctx context.Context, message *pbMsg.MsgDataToMQ, pushToUserID string) {
 	log.Info(message.OperationID, utils.GetSelfFuncName(), "msg ", message.String(), pushToUserID)
 	rpcPushMsg := pbPush.PushMsgReq{OperationID: message.OperationID, MsgData: message.MsgData, PushToUserID: pushToUserID}
 	mqPushMsg := pbMsg.PushMsgDataToMQ{OperationID: message.OperationID, MsgData: message.MsgData, PushToUserID: pushToUserID}
-	pid, offset, err := producer.SendMessage(&mqPushMsg, mqPushMsg.PushToUserID, rpcPushMsg.OperationID)
+	pid, offset, err := och.producerToPush.SendMessage(&mqPushMsg, mqPushMsg.PushToUserID, rpcPushMsg.OperationID)
 	if err != nil {
 		log.Error(mqPushMsg.OperationID, "kafka send failed", "send data", message.String(), "pid", pid, "offset", offset, "err", err.Error())
 	}
 	return
 }

-func sendMessageToModifyMQ(aggregationID string, triggerID string, messages []*pbMsg.MsgDataToMQ) {
+func (och *OnlineHistoryRedisConsumerHandler) sendMessageToModifyMQ(ctx context.Context, aggregationID string, triggerID string, messages []*pbMsg.MsgDataToMQ) {
 	if len(messages) > 0 {
-		pid, offset, err := producerToModify.SendMessage(&pbMsg.MsgDataToModifyByMQ{AggregationID: aggregationID, MessageList: messages, TriggerID: triggerID}, aggregationID, triggerID)
+		pid, offset, err := och.producerToModify.SendMessage(&pbMsg.MsgDataToModifyByMQ{AggregationID: aggregationID, MessageList: messages, TriggerID: triggerID}, aggregationID, triggerID)
 		if err != nil {
 			log.Error(triggerID, "kafka send failed", "send data", len(messages), "pid", pid, "offset", offset, "err", err.Error(), "key", aggregationID)
-		} else {
-			//	log.NewWarn(m.OperationID, "sendMsgToKafka   client msgID ", m.MsgData.ClientMsgID)
 		}
 	}
 }

-// String hashes a string to a unique hashcode.
-//
-// crc32 returns a uint32, but for our use we need
-// and non negative integer. Here we cast to an integer
-// and invert it if the result is negative.
-func getHashCode(s string) uint32 {
-	return crc32.ChecksumIEEE([]byte(s))
+func (och *OnlineHistoryRedisConsumerHandler) SendMessageToMongoCH(ctx context.Context, aggregationID string, triggerID string, messages []*pbMsg.MsgDataToMQ, lastSeq uint64) {
+	if len(messages) > 0 {
+		pid, offset, err := och.producerToMongo.SendMessage(&pbMsg.MsgDataToMongoByMQ{LastSeq: lastSeq, AggregationID: aggregationID, MessageList: messages, TriggerID: triggerID}, aggregationID, triggerID)
+		if err != nil {
+			log.Error(triggerID, "kafka send failed", "send data", len(messages), "pid", pid, "offset", offset, "err", err.Error(), "key", aggregationID)
+		}
+	}
 }
@@ -3,28 +3,29 @@ package msgtransfer
 import (
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
-	"Open_IM/pkg/common/db"
+	"Open_IM/pkg/common/db/cache"
+	"Open_IM/pkg/common/db/controller"
 	kfk "Open_IM/pkg/common/kafka"
 	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/tracelog"
 	pbMsg "Open_IM/pkg/proto/msg"
-	sdkws "Open_IM/pkg/proto/sdkws"
+	"Open_IM/pkg/proto/sdkws"
 	"Open_IM/pkg/utils"
+	"context"
 	"github.com/Shopify/sarama"
 	"github.com/golang/protobuf/proto"
 )

 type OnlineHistoryMongoConsumerHandler struct {
-	msgHandle            map[string]fcb
 	historyConsumerGroup *kfk.MConsumerGroup
+	msgInterface         controller.MsgInterface
+	cache                cache.Cache
 }

 func (mc *OnlineHistoryMongoConsumerHandler) Init() {
-	mc.msgHandle = make(map[string]fcb)
-	mc.msgHandle[config.Config.Kafka.MsgToMongo.Topic] = mc.handleChatWs2Mongo
 	mc.historyConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
 		OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.MsgToMongo.Topic},
 		config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMongo)
-
 }
 func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
 	msg := cMsg.Value
@@ -35,14 +36,17 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(cMsg *sarama.Con
 		return
 	}
 	log.Info(msgFromMQ.TriggerID, "BatchInsertChat2DB userID: ", msgFromMQ.AggregationID, "msgFromMQ.LastSeq: ", msgFromMQ.LastSeq)
-	err = db.DB.BatchInsertChat2DB(msgFromMQ.AggregationID, msgFromMQ.MessageList, msgFromMQ.TriggerID, msgFromMQ.LastSeq)
+	ctx := context.Background()
+	tracelog.SetOperationID(ctx, msgFromMQ.TriggerID)
+	//err = db.DB.BatchInsertChat2DB(msgFromMQ.AggregationID, msgFromMQ.MessageList, msgFromMQ.TriggerID, msgFromMQ.LastSeq)
+	err = mc.msgInterface.BatchInsertChat2DB(ctx, msgFromMQ.AggregationID, msgFromMQ.MessageList, msgFromMQ.LastSeq)
 	if err != nil {
 		log.NewError(msgFromMQ.TriggerID, "single data insert to mongo err", err.Error(), msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.TriggerID)
-	} else {
-		err = db.DB.DeleteMessageFromCache(msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.GetTriggerID())
-		if err != nil {
-			log.NewError(msgFromMQ.TriggerID, "remove cache msg from redis  err", err.Error(), msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.TriggerID)
-		}
+	}
+	//err = db.DB.DeleteMessageFromCache(msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.GetTriggerID())
+	err = mc.msgInterface.DeleteMessageFromCache(ctx, msgFromMQ.AggregationID, msgFromMQ.MessageList)
+	if err != nil {
+		log.NewError(msgFromMQ.TriggerID, "remove cache msg from redis err", err.Error(), msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.TriggerID)
 	}
 	for _, v := range msgFromMQ.MessageList {
 		if v.MsgData.ContentType == constant.DeleteMessageNotification {
@@ -58,23 +62,23 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(cMsg *sarama.Con
 				log.NewError(msgFromMQ.TriggerID, "deleteMessageTips unmarshal err:", err.Error(), v.String())
 				continue
 			}
-			if unexistSeqList, err := db.DB.DelMsgBySeqList(DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID); err != nil {
-				log.NewError(v.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqList args: ", DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID, err.Error(), unexistSeqList)
+			if totalUnExistSeqs, err := mc.msgInterface.DelMsgBySeqs(ctx, DeleteMessageTips.UserID, DeleteMessageTips.SeqList); err != nil {
+				log.NewError(v.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqs args: ", DeleteMessageTips.UserID, DeleteMessageTips.SeqList, "error:", err.Error(), "totalUnExistSeqs: ", totalUnExistSeqs)
 			}
+
 		}
 	}
 }

 func (OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
 func (OnlineHistoryMongoConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
-
 func (mc *OnlineHistoryMongoConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
 	claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
 	log.NewDebug("", "online new session msg come", claim.HighWaterMarkOffset(), claim.Topic(), claim.Partition())
 	for msg := range claim.Messages() {
 		log.NewDebug("", "kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
 		if len(msg.Value) != 0 {
-			mc.msgHandle[msg.Topic](msg, string(msg.Key), sess)
+			mc.handleChatWs2Mongo(msg, string(msg.Key), sess)
 		} else {
 			log.Error("", "mongo msg get from kafka but is nil", msg.Key)
 		}
@@ -9,7 +9,7 @@ package msgtransfer
 import (
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
-	"Open_IM/pkg/common/db/mysql_model/im_mysql_msg_model"
+	"Open_IM/pkg/common/db/controller"
 	kfk "Open_IM/pkg/common/kafka"
 	"Open_IM/pkg/common/log"
 	pbMsg "Open_IM/pkg/proto/msg"
@@ -17,33 +17,17 @@ import (

 	"github.com/Shopify/sarama"
 	"github.com/golang/protobuf/proto"
-
-	promePkg "Open_IM/pkg/common/prometheus"
 )

 type PersistentConsumerHandler struct {
-	msgHandle               map[string]fcb
 	persistentConsumerGroup *kfk.MConsumerGroup
+	chatLogInterface        controller.ChatLogInterface
 }

 func (pc *PersistentConsumerHandler) Init() {
-	pc.msgHandle = make(map[string]fcb)
-	pc.msgHandle[config.Config.Kafka.Ws2mschat.Topic] = pc.handleChatWs2Mysql
 	pc.persistentConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
 		OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ws2mschat.Topic},
 		config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMySql)
-
-}
-
-func initPrometheus() {
-	promePkg.NewSeqGetSuccessCounter()
-	promePkg.NewSeqGetFailedCounter()
-	promePkg.NewSeqSetSuccessCounter()
-	promePkg.NewSeqSetFailedCounter()
-	promePkg.NewMsgInsertRedisSuccessCounter()
-	promePkg.NewMsgInsertRedisFailedCounter()
-	promePkg.NewMsgInsertMongoSuccessCounter()
-	promePkg.NewMsgInsertMongoFailedCounter()
 }

 func (pc *PersistentConsumerHandler) handleChatWs2Mysql(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
@@ -75,7 +59,7 @@ func (pc *PersistentConsumerHandler) handleChatWs2Mysql(cMsg *sarama.ConsumerMes
 		}
 		if tag {
 			log.NewInfo(msgFromMQ.OperationID, "msg_transfer msg persisting", string(msg))
-			if err = im_mysql_msg_model.InsertMessageToChatLog(msgFromMQ); err != nil {
+			if err = pc.chatLogInterface.CreateChatLog(msgFromMQ); err != nil {
 				log.NewError(msgFromMQ.OperationID, "Message insert failed", "err", err.Error(), "msg", msgFromMQ.String())
 				return
 			}
@@ -90,7 +74,7 @@ func (pc *PersistentConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSessi
 	for msg := range claim.Messages() {
 		log.NewDebug("", "kafka get info to mysql", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
 		if len(msg.Value) != 0 {
-			pc.msgHandle[msg.Topic](msg, string(msg.Key), sess)
+			pc.handleChatWs2Mysql(msg, string(msg.Key), sess)
 		} else {
 			log.Error("", "msg get from kafka but is nil", msg.Key)
 		}
@@ -98,15 +82,3 @@ func (pc *PersistentConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSessi
 	}
 	return nil
 }
-
-1. 请求1  group Rpc             2. 请求2 发消息 sendMsg rpc
-1 更改数据库
-
-2. 删除哈希缓存
-								检测到哈希变了， 群成员还没来得及删除，有问题
-3. 删除群成员缓存
-
-4. 删除对应群成员加群缓存
-
-5. 删除数量缓存
-
@@ -15,7 +15,7 @@ import (
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/kafka"
-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/statistics"
 	"fmt"
 )
@@ -53,15 +53,15 @@ func init() {
 }

 func initPrometheus() {
-	promePkg.NewMsgOfflinePushSuccessCounter()
-	promePkg.NewMsgOfflinePushFailedCounter()
+	prome.NewMsgOfflinePushSuccessCounter()
+	prome.NewMsgOfflinePushFailedCounter()
 }

 func Run(promethuesPort int) {
 	go rpcServer.run()
 	go pushCh.pushConsumerGroup.RegisterHandleAndConsumer(&pushCh)
 	go func() {
-		err := promePkg.StartPromeSrv(promethuesPort)
+		err := prome.StartPromeSrv(promethuesPort)
 		if err != nil {
 			panic(err)
 		}
@@ -5,7 +5,7 @@ import (
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/db"
 	"Open_IM/pkg/common/log"
-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/getcdv3"
 	pbPush "Open_IM/pkg/proto/push"
 	"Open_IM/pkg/utils"
@@ -47,11 +47,11 @@ func (r *RPCServer) run() {
 	defer listener.Close()
 	var grpcOpts []grpc.ServerOption
 	if config.Config.Prometheus.Enable {
-		promePkg.NewGrpcRequestCounter()
-		promePkg.NewGrpcRequestFailedCounter()
-		promePkg.NewGrpcRequestSuccessCounter()
+		prome.NewGrpcRequestCounter()
+		prome.NewGrpcRequestFailedCounter()
+		prome.NewGrpcRequestSuccessCounter()
 		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
+			// grpc.UnaryInterceptor(prome.UnaryServerInterceptorProme),
 			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
 			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
 		}...)
@@ -20,7 +20,7 @@ import (
 	"context"
 	"strings"

-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"

 	"github.com/golang/protobuf/proto"
 )
@@ -144,10 +144,10 @@ func MsgToUser(pushMsg *pbPush.PushMsgReq) {
 		}
 		pushResult, err := offlinePusher.Push(UIDList, title, detailContent, pushMsg.OperationID, opts)
 		if err != nil {
-			promePkg.PromeInc(promePkg.MsgOfflinePushFailedCounter)
+			prome.PromeInc(prome.MsgOfflinePushFailedCounter)
 			log.NewError(pushMsg.OperationID, "offline push error", pushMsg.String(), err.Error())
 		} else {
-			promePkg.PromeInc(promePkg.MsgOfflinePushSuccessCounter)
+			prome.PromeInc(prome.MsgOfflinePushSuccessCounter)
 			log.NewDebug(pushMsg.OperationID, "offline push return result is ", pushResult, pushMsg.MsgData)
 		}
 	}
@@ -267,10 +267,10 @@ func MsgToSuperGroupUser(pushMsg *pbPush.PushMsgReq) {
 			}
 			pushResult, err := offlinePusher.Push(needOfflinePushUserIDList, title, detailContent, pushMsg.OperationID, opts)
 			if err != nil {
-				promePkg.PromeInc(promePkg.MsgOfflinePushFailedCounter)
+				prome.PromeInc(prome.MsgOfflinePushFailedCounter)
 				log.NewError(pushMsg.OperationID, "offline push error", pushMsg.String(), err.Error())
 			} else {
-				promePkg.PromeInc(promePkg.MsgOfflinePushSuccessCounter)
+				prome.PromeInc(prome.MsgOfflinePushSuccessCounter)
 				log.NewDebug(pushMsg.OperationID, "offline push return result is ", pushResult, pushMsg.MsgData)
 			}
 		}
@@ -8,7 +8,7 @@ import (
 	"Open_IM/pkg/common/db/cache"
 	"Open_IM/pkg/common/db/controller"
 	"Open_IM/pkg/common/log"
-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/common/tokenverify"
 	"Open_IM/pkg/common/tracelog"
 	pbAuth "Open_IM/pkg/proto/auth"
@@ -42,13 +42,13 @@ func (s *rpcAuth) Run() {
 	log.NewInfo(operationID, "listen network success ", listener, address)
 	var grpcOpts []grpc.ServerOption
 	if config.Config.Prometheus.Enable {
-		promePkg.NewGrpcRequestCounter()
-		promePkg.NewGrpcRequestFailedCounter()
-		promePkg.NewGrpcRequestSuccessCounter()
-		promePkg.NewUserRegisterCounter()
-		promePkg.NewUserLoginCounter()
+		prome.NewGrpcRequestCounter()
+		prome.NewGrpcRequestFailedCounter()
+		prome.NewGrpcRequestSuccessCounter()
+		prome.NewUserRegisterCounter()
+		prome.NewUserLoginCounter()
 		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
+			// grpc.UnaryInterceptor(prome.UnaryServerInterceptorProme),
 			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
 			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
 		}...)
@@ -6,14 +6,11 @@ import (
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/db/cache"
 	"Open_IM/pkg/common/db/controller"
-	"Open_IM/pkg/common/db/relation"
-	"Open_IM/pkg/common/db/table"
-	"Open_IM/pkg/common/db/unrelation"
+	relationTb "Open_IM/pkg/common/db/relation"
+	unrealationTb "Open_IM/pkg/common/db/unrelation"
 	"Open_IM/pkg/common/log"
-	promePkg "Open_IM/pkg/common/prometheus"
-	"Open_IM/pkg/getcdv3"
+	"Open_IM/pkg/common/prome"
 	pbConversation "Open_IM/pkg/proto/conversation"
-	pbUser "Open_IM/pkg/proto/user"
 	"Open_IM/pkg/utils"
 	"context"
 	"github.com/dtm-labs/rockscache"
@@ -95,11 +92,11 @@ func (c *conversationServer) Run() {
 	//grpc server
 	var grpcOpts []grpc.ServerOption
 	if config.Config.Prometheus.Enable {
-		promePkg.NewGrpcRequestCounter()
-		promePkg.NewGrpcRequestFailedCounter()
-		promePkg.NewGrpcRequestSuccessCounter()
+		prome.NewGrpcRequestCounter()
+		prome.NewGrpcRequestFailedCounter()
+		prome.NewGrpcRequestSuccessCounter()
 		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
+			// grpc.UnaryInterceptor(prome.UnaryServerInterceptorProme),
 			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
 			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
 		}...)
@@ -12,7 +12,7 @@ import (
 	relationTb "Open_IM/pkg/common/db/table/relation"
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/common/middleware"
-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/common/tokenverify"
 	"Open_IM/pkg/common/tracelog"
 	pbFriend "Open_IM/pkg/proto/friend"
@@ -75,11 +75,11 @@ func (s *friendServer) Run() {
 	var grpcOpts []grpc.ServerOption
 	grpcOpts = append(grpcOpts, grpc.UnaryInterceptor(middleware.RpcServerInterceptor))
 	if config.Config.Prometheus.Enable {
-		promePkg.NewGrpcRequestCounter()
-		promePkg.NewGrpcRequestFailedCounter()
-		promePkg.NewGrpcRequestSuccessCounter()
+		prome.NewGrpcRequestCounter()
+		prome.NewGrpcRequestFailedCounter()
+		prome.NewGrpcRequestSuccessCounter()
 		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
+			// grpc.UnaryInterceptor(prome.UnaryServerInterceptorProme),
 			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
 			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
 		}...)
@@ -1,7 +1,7 @@
 package group

 import (
-	cbApi "Open_IM/pkg/callback_struct"
+	"Open_IM/pkg/callbackstruct"
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/db/table/relation"
@@ -12,7 +12,7 @@ import (
 	"Open_IM/pkg/common/db/unrelation"
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/common/middleware"
-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/common/tokenverify"
 	"Open_IM/pkg/common/tracelog"
 	discoveryRegistry "Open_IM/pkg/discoveryregistry"
@@ -70,9 +70,7 @@ func NewGroupServer(port int) *groupServer {
 	if err != nil {
 		panic(err.Error())
 	}
-
 	//conns, err := g.registerCenter.GetConns(config.Config.RpcRegisterName.OpenImConversationName)
-
 	g.GroupInterface = controller.NewGroupInterface(mysql.GormConn(), redis.GetClient(), mongo.GetClient())
 	return &g
 }
@@ -98,11 +96,11 @@ func (s *groupServer) Run() {
 		grpc.UnaryInterceptor(middleware.RpcServerInterceptor),
 	}
 	if config.Config.Prometheus.Enable {
-		promePkg.NewGrpcRequestCounter()
-		promePkg.NewGrpcRequestFailedCounter()
-		promePkg.NewGrpcRequestSuccessCounter()
+		prome.NewGrpcRequestCounter()
+		prome.NewGrpcRequestFailedCounter()
+		prome.NewGrpcRequestSuccessCounter()
 		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
+			// grpc.UnaryInterceptor(prome.UnaryServerInterceptorProme),
 			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
 			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
 		}...)
@@ -9,7 +9,7 @@ import (
 	"Open_IM/pkg/common/log"
 	sdkws "Open_IM/pkg/proto/sdkws"

-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"
 )

 func (rpc *rpcChat) GetMaxAndMinSeq(_ context.Context, in *sdkws.GetMaxAndMinSeqReq) (*sdkws.GetMaxAndMinSeqResp, error) {
@@ -53,25 +53,25 @@ func (rpc *rpcChat) PullMessageBySeqList(_ context.Context, in *sdkws.PullMessag
 	redisMsgList, failedSeqList, err := commonDB.DB.GetMessageListBySeq(in.UserID, in.SeqList, in.OperationID)
 	if err != nil {
 		if err != go_redis.Nil {
-			promePkg.PromeAdd(promePkg.MsgPullFromRedisFailedCounter, len(failedSeqList))
+			prome.PromeAdd(prome.MsgPullFromRedisFailedCounter, len(failedSeqList))
 			log.Error(in.OperationID, "get message from redis exception", err.Error(), failedSeqList)
 		} else {
 			log.Debug(in.OperationID, "get message from redis is nil", failedSeqList)
 		}
-		msgList, err1 := commonDB.DB.GetMsgBySeqListMongo2(in.UserID, failedSeqList, in.OperationID)
+		msgList, err1 := commonDB.DB.GetMsgBySeqs(in.UserID, failedSeqList, in.OperationID)
 		if err1 != nil {
-			promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList))
+			prome.PromeAdd(prome.MsgPullFromMongoFailedCounter, len(failedSeqList))
 			log.Error(in.OperationID, "PullMessageBySeqList data error", in.String(), err1.Error())
 			resp.ErrCode = 201
 			resp.ErrMsg = err1.Error()
 			return resp, nil
 		} else {
-			promePkg.PromeAdd(promePkg.MsgPullFromMongoSuccessCounter, len(msgList))
+			prome.PromeAdd(prome.MsgPullFromMongoSuccessCounter, len(msgList))
 			redisMsgList = append(redisMsgList, msgList...)
 			resp.List = redisMsgList
 		}
 	} else {
-		promePkg.PromeAdd(promePkg.MsgPullFromRedisSuccessCounter, len(redisMsgList))
+		prome.PromeAdd(prome.MsgPullFromRedisSuccessCounter, len(redisMsgList))
 		resp.List = redisMsgList
 	}

@@ -80,26 +80,26 @@ func (rpc *rpcChat) PullMessageBySeqList(_ context.Context, in *sdkws.PullMessag
 		redisMsgList, failedSeqList, err := commonDB.DB.GetMessageListBySeq(k, v.SeqList, in.OperationID)
 		if err != nil {
 			if err != go_redis.Nil {
-				promePkg.PromeAdd(promePkg.MsgPullFromRedisFailedCounter, len(failedSeqList))
+				prome.PromeAdd(prome.MsgPullFromRedisFailedCounter, len(failedSeqList))
 				log.Error(in.OperationID, "get message from redis exception", err.Error(), failedSeqList)
 			} else {
 				log.Debug(in.OperationID, "get message from redis is nil", failedSeqList)
 			}
-			msgList, err1 := commonDB.DB.GetSuperGroupMsgBySeqListMongo(k, failedSeqList, in.OperationID)
+			msgList, err1 := commonDB.DB.GetSuperGroupMsgBySeqs(k, failedSeqList, in.OperationID)
 			if err1 != nil {
-				promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList))
+				prome.PromeAdd(prome.MsgPullFromMongoFailedCounter, len(failedSeqList))
 				log.Error(in.OperationID, "PullMessageBySeqList data error", in.String(), err1.Error())
 				resp.ErrCode = 201
 				resp.ErrMsg = err1.Error()
 				return resp, nil
 			} else {
-				promePkg.PromeAdd(promePkg.MsgPullFromMongoSuccessCounter, len(msgList))
+				prome.PromeAdd(prome.MsgPullFromMongoSuccessCounter, len(msgList))
 				redisMsgList = append(redisMsgList, msgList...)
 				x.MsgDataList = redisMsgList
 				m[k] = x
 			}
 		} else {
-			promePkg.PromeAdd(promePkg.MsgPullFromRedisSuccessCounter, len(redisMsgList))
+			prome.PromeAdd(prome.MsgPullFromRedisSuccessCounter, len(redisMsgList))
 			x.MsgDataList = redisMsgList
 			m[k] = x
 		}
@@ -3,7 +3,7 @@ package msg
 import (
 	commonDB "Open_IM/pkg/common/db"
 	"Open_IM/pkg/common/log"
-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/proto/msg"
 	"Open_IM/pkg/utils"
 	"context"
@@ -16,20 +16,20 @@ func (rpc *rpcChat) GetSuperGroupMsg(context context.Context, req *msg.GetSuperG
 	redisMsgList, failedSeqList, err := commonDB.DB.GetMessageListBySeq(req.GroupID, []uint32{req.Seq}, req.OperationID)
 	if err != nil {
 		if err != go_redis.Nil {
-			promePkg.PromeAdd(promePkg.MsgPullFromRedisFailedCounter, len(failedSeqList))
+			prome.PromeAdd(prome.MsgPullFromRedisFailedCounter, len(failedSeqList))
 			log.Error(req.OperationID, "get message from redis exception", err.Error(), failedSeqList)
 		} else {
 			log.Debug(req.OperationID, "get message from redis is nil", failedSeqList)
 		}
-		msgList, err1 := commonDB.DB.GetSuperGroupMsgBySeqListMongo(req.GroupID, failedSeqList, req.OperationID)
+		msgList, err1 := commonDB.DB.GetSuperGroupMsgBySeqs(req.GroupID, failedSeqList, req.OperationID)
 		if err1 != nil {
-			promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList))
+			prome.PromeAdd(prome.MsgPullFromMongoFailedCounter, len(failedSeqList))
 			log.Error(req.OperationID, "GetSuperGroupMsg data error", req.String(), err.Error())
 			resp.ErrCode = 201
 			resp.ErrMsg = err.Error()
 			return resp, nil
 		} else {
-			promePkg.PromeAdd(promePkg.MsgPullFromMongoSuccessCounter, len(msgList))
+			prome.PromeAdd(prome.MsgPullFromMongoSuccessCounter, len(msgList))
 			redisMsgList = append(redisMsgList, msgList...)
 			for _, m := range msgList {
 				resp.MsgData = m
@@ -37,7 +37,7 @@ func (rpc *rpcChat) GetSuperGroupMsg(context context.Context, req *msg.GetSuperG

 		}
 	} else {
-		promePkg.PromeAdd(promePkg.MsgPullFromRedisSuccessCounter, len(redisMsgList))
+		prome.PromeAdd(prome.MsgPullFromRedisSuccessCounter, len(redisMsgList))
 		for _, m := range redisMsgList {
 			resp.MsgData = m
 		}
@@ -6,7 +6,7 @@ import (
 	"Open_IM/pkg/common/db"
 	"Open_IM/pkg/common/kafka"
 	"Open_IM/pkg/common/log"
-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/proto/msg"
 	"Open_IM/pkg/utils"
 	"github.com/OpenIMSDK/getcdv3"
@@ -66,21 +66,21 @@ func (rpc *rpcChat) initPrometheus() {
 	//	Name: "send_msg_failed",
 	//	Help: "The number of send msg failed",
 	//})
-	promePkg.NewMsgPullFromRedisSuccessCounter()
-	promePkg.NewMsgPullFromRedisFailedCounter()
-	promePkg.NewMsgPullFromMongoSuccessCounter()
-	promePkg.NewMsgPullFromMongoFailedCounter()
+	prome.NewMsgPullFromRedisSuccessCounter()
+	prome.NewMsgPullFromRedisFailedCounter()
+	prome.NewMsgPullFromMongoSuccessCounter()
+	prome.NewMsgPullFromMongoFailedCounter()

-	promePkg.NewSingleChatMsgRecvSuccessCounter()
-	promePkg.NewGroupChatMsgRecvSuccessCounter()
-	promePkg.NewWorkSuperGroupChatMsgRecvSuccessCounter()
+	prome.NewSingleChatMsgRecvSuccessCounter()
+	prome.NewGroupChatMsgRecvSuccessCounter()
+	prome.NewWorkSuperGroupChatMsgRecvSuccessCounter()

-	promePkg.NewSingleChatMsgProcessSuccessCounter()
-	promePkg.NewSingleChatMsgProcessFailedCounter()
-	promePkg.NewGroupChatMsgProcessSuccessCounter()
-	promePkg.NewGroupChatMsgProcessFailedCounter()
-	promePkg.NewWorkSuperGroupChatMsgProcessSuccessCounter()
-	promePkg.NewWorkSuperGroupChatMsgProcessFailedCounter()
+	prome.NewSingleChatMsgProcessSuccessCounter()
+	prome.NewSingleChatMsgProcessFailedCounter()
+	prome.NewGroupChatMsgProcessSuccessCounter()
+	prome.NewGroupChatMsgProcessFailedCounter()
+	prome.NewWorkSuperGroupChatMsgProcessSuccessCounter()
+	prome.NewWorkSuperGroupChatMsgProcessFailedCounter()
 }

 func (rpc *rpcChat) Run() {
@@ -104,11 +104,11 @@ func (rpc *rpcChat) Run() {
 		grpc.MaxSendMsgSize(sendSize),
 	}
 	if config.Config.Prometheus.Enable {
-		promePkg.NewGrpcRequestCounter()
-		promePkg.NewGrpcRequestFailedCounter()
-		promePkg.NewGrpcRequestSuccessCounter()
+		prome.NewGrpcRequestCounter()
+		prome.NewGrpcRequestFailedCounter()
+		prome.NewGrpcRequestSuccessCounter()
 		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
+			// grpc.UnaryInterceptor(prome.UnaryServerInterceptorProme),
 			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
 			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
 		}...)
@@ -146,9 +146,9 @@ func (rpc *rpcChat) runCh() {
 		case msg := <-rpc.delMsgCh:
 			log.NewInfo(msg.OperationID, utils.GetSelfFuncName(), "delmsgch recv new: ", msg)
 			db.DB.DelMsgFromCache(msg.UserID, msg.SeqList, msg.OperationID)
-			unexistSeqList, err := db.DB.DelMsgBySeqList(msg.UserID, msg.SeqList, msg.OperationID)
+			unexistSeqList, err := db.DB.DelMsgBySeqs(msg.UserID, msg.SeqList, msg.OperationID)
 			if err != nil {
-				log.NewError(msg.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqList args: ", msg.UserID, msg.SeqList, msg.OperationID, err.Error())
+				log.NewError(msg.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqs args: ", msg.UserID, msg.SeqList, msg.OperationID, err.Error())
 				continue
 			}
 			if len(unexistSeqList) > 0 {
@@ -23,7 +23,7 @@ import (
 	"sync"
 	"time"

-	promePkg "Open_IM/pkg/common/prometheus"
+	prome "Open_IM/pkg/common/prometheus"
 	go_redis "github.com/go-redis/redis/v8"
 	"github.com/golang/protobuf/proto"
 )
@@ -369,7 +369,7 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 	}
 	switch pb.MsgData.SessionType {
 	case constant.SingleChatType:
-		promePkg.PromeInc(promePkg.SingleChatMsgRecvSuccessCounter)
+		prome.PromeInc(prome.SingleChatMsgRecvSuccessCounter)
 		// callback
 		t1 = time.Now()
 		callbackResp := callbackBeforeSendSingleMsg(pb)
@@ -382,7 +382,7 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 				callbackResp.ErrCode = 201
 			}
 			log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSingleMsg result", "end rpc and return", callbackResp)
-			promePkg.PromeInc(promePkg.SingleChatMsgProcessFailedCounter)
+			prome.PromeInc(prome.SingleChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, int32(callbackResp.ErrCode), callbackResp.ErrMsg, "", 0)
 		}
 		t1 = time.Now()
@@ -402,7 +402,7 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 			log.Info(pb.OperationID, "sendMsgToWriter ", " cost time: ", time.Since(t1))
 			if err1 != nil {
 				log.NewError(msgToMQSingle.OperationID, "kafka send msg err :RecvID", msgToMQSingle.MsgData.RecvID, msgToMQSingle.String(), err1.Error())
-				promePkg.PromeInc(promePkg.SingleChatMsgProcessFailedCounter)
+				prome.PromeInc(prome.SingleChatMsgProcessFailedCounter)
 				return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
 			}
 		}
@@ -412,7 +412,7 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 			log.Info(pb.OperationID, "sendMsgToWriter ", " cost time: ", time.Since(t1))
 			if err2 != nil {
 				log.NewError(msgToMQSingle.OperationID, "kafka send msg err:SendID", msgToMQSingle.MsgData.SendID, msgToMQSingle.String())
-				promePkg.PromeInc(promePkg.SingleChatMsgProcessFailedCounter)
+				prome.PromeInc(prome.SingleChatMsgProcessFailedCounter)
 				return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
 			}
 		}
@@ -423,11 +423,11 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 		if callbackResp.ErrCode != 0 {
 			log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendSingleMsg resp: ", callbackResp)
 		}
-		promePkg.PromeInc(promePkg.SingleChatMsgProcessSuccessCounter)
+		prome.PromeInc(prome.SingleChatMsgProcessSuccessCounter)
 		return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)
 	case constant.GroupChatType:
 		// callback
-		promePkg.PromeInc(promePkg.GroupChatMsgRecvSuccessCounter)
+		prome.PromeInc(prome.GroupChatMsgRecvSuccessCounter)
 		callbackResp := callbackBeforeSendGroupMsg(pb)
 		if callbackResp.ErrCode != 0 {
 			log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendGroupMsg resp:", callbackResp)
@@ -437,12 +437,12 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 				callbackResp.ErrCode = 201
 			}
 			log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSingleMsg result", "end rpc and return", callbackResp)
-			promePkg.PromeInc(promePkg.GroupChatMsgProcessFailedCounter)
+			prome.PromeInc(prome.GroupChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, int32(callbackResp.ErrCode), callbackResp.ErrMsg, "", 0)
 		}
 		var memberUserIDList []string
 		if flag, errCode, errMsg, memberUserIDList = rpc.messageVerification(ctx, pb); !flag {
-			promePkg.PromeInc(promePkg.GroupChatMsgProcessFailedCounter)
+			prome.PromeInc(prome.GroupChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, errCode, errMsg, "", 0)
 		}
 		log.Debug(pb.OperationID, "GetGroupAllMember userID list", memberUserIDList, "len: ", len(memberUserIDList))
@@ -506,7 +506,7 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 		}
 		if !sendTag {
 			log.NewWarn(pb.OperationID, "send tag is ", sendTag)
-			promePkg.PromeInc(promePkg.GroupChatMsgProcessFailedCounter)
+			prome.PromeInc(prome.GroupChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
 		} else {
 			if pb.MsgData.ContentType == constant.AtText {
@@ -571,7 +571,7 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 				}()
 			}
 			log.Debug(pb.OperationID, "send msg cost time3 ", time.Since(t1), pb.MsgData.ClientMsgID)
-			promePkg.PromeInc(promePkg.GroupChatMsgProcessSuccessCounter)
+			prome.PromeInc(prome.GroupChatMsgProcessSuccessCounter)
 			return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)
 		}
 	case constant.NotificationChatType:
@@ -595,7 +595,7 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 		log.Debug(pb.OperationID, "send msg cost time ", time.Since(t1), pb.MsgData.ClientMsgID)
 		return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)
 	case constant.SuperGroupChatType:
-		promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgRecvSuccessCounter)
+		prome.PromeInc(prome.WorkSuperGroupChatMsgRecvSuccessCounter)
 		// callback
 		callbackResp := callbackBeforeSendGroupMsg(pb)
 		if callbackResp.ErrCode != 0 {
@@ -605,12 +605,12 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 			if callbackResp.ErrCode == 0 {
 				callbackResp.ErrCode = 201
 			}
-			promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
+			prome.PromeInc(prome.WorkSuperGroupChatMsgProcessFailedCounter)
 			log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSuperGroupMsg result", "end rpc and return", callbackResp)
 			return returnMsg(&replay, pb, int32(callbackResp.ErrCode), callbackResp.ErrMsg, "", 0)
 		}
 		if flag, errCode, errMsg, _ = rpc.messageVerification(ctx, pb); !flag {
-			promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
+			prome.PromeInc(prome.WorkSuperGroupChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, errCode, errMsg, "", 0)
 		}
 		msgToMQSingle.MsgData = pb.MsgData
@@ -618,7 +618,7 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 		err1 := rpc.sendMsgToWriter(ctx, &msgToMQSingle, msgToMQSingle.MsgData.GroupID, constant.OnlineStatus)
 		if err1 != nil {
 			log.NewError(msgToMQSingle.OperationID, "kafka send msg err:RecvID", msgToMQSingle.MsgData.RecvID, msgToMQSingle.String())
-			promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
+			prome.PromeInc(prome.WorkSuperGroupChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
 		}
 		// callback
@@ -626,7 +626,7 @@ func (rpc *rpcChat) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat
 		if callbackResp.ErrCode != 0 {
 			log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendSuperGroupMsg resp: ", callbackResp)
 		}
-		promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessSuccessCounter)
+		prome.PromeInc(prome.WorkSuperGroupChatMsgProcessSuccessCounter)
 		return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)

 	default:
@@ -1 +0,0 @@
-package user
@@ -3,6 +3,7 @@ package user
 import (
 	"Open_IM/internal/common/convert"
 	"Open_IM/internal/common/rpc_server"
+	"Open_IM/internal/common/rpcserver"
 	chat "Open_IM/internal/rpc/msg"
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
@@ -65,7 +66,7 @@ func (s *userServer) Run() {
 		prome.NewGrpcRequestFailedCounter()
 		prome.NewGrpcRequestSuccessCounter()
 		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
+			// grpc.UnaryInterceptor(prome.UnaryServerInterceptorProme),
 			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
 			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
 		}...)