Merge 8cd57d68535f83c6495423e777f6a30c88cbe0f1 into 812c1e41271750306c3b4705c555a0bc7913aaa4

internal/msgtransfer/callback.go

@@ -1,124 +0,0 @@
-package msgtransfer
-
-import (
-	"context"
-	"encoding/base64"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
-	"github.com/openimsdk/protocol/constant"
-	"github.com/openimsdk/protocol/sdkws"
-	"github.com/openimsdk/tools/mcontext"
-	"github.com/openimsdk/tools/utils/datautil"
-	"github.com/openimsdk/tools/utils/stringutil"
-	"google.golang.org/protobuf/proto"
-
-	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
-)
-
-func toCommonCallback(ctx context.Context, msg *sdkws.MsgData, command string) cbapi.CommonCallbackReq {
-	return cbapi.CommonCallbackReq{
-		SendID:           msg.SendID,
-		ServerMsgID:      msg.ServerMsgID,
-		CallbackCommand:  command,
-		ClientMsgID:      msg.ClientMsgID,
-		OperationID:      mcontext.GetOperationID(ctx),
-		SenderPlatformID: msg.SenderPlatformID,
-		SenderNickname:   msg.SenderNickname,
-		SessionType:      msg.SessionType,
-		MsgFrom:          msg.MsgFrom,
-		ContentType:      msg.ContentType,
-		Status:           msg.Status,
-		SendTime:         msg.SendTime,
-		CreateTime:       msg.CreateTime,
-		AtUserIDList:     msg.AtUserIDList,
-		SenderFaceURL:    msg.SenderFaceURL,
-		Content:          GetContent(msg),
-		Seq:              uint32(msg.Seq),
-		Ex:               msg.Ex,
-	}
-}
-
-func GetContent(msg *sdkws.MsgData) string {
-	if msg.ContentType >= constant.NotificationBegin && msg.ContentType <= constant.NotificationEnd {
-		var tips sdkws.TipsComm
-		_ = proto.Unmarshal(msg.Content, &tips)
-		content := tips.JsonDetail
-		return content
-	} else {
-		return string(msg.Content)
-	}
-}
-
-func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) {
-	if msg.ContentType == constant.Typing {
-		return
-	}
-	if !filterAfterMsg(msg, after) {
-		return
-	}
-	cbReq := &cbapi.CallbackAfterSendSingleMsgReq{
-		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand),
-		RecvID:            msg.RecvID,
-	}
-	mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg))
-}
-
-func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) {
-	if msg.ContentType == constant.Typing {
-		return
-	}
-	if !filterAfterMsg(msg, after) {
-		return
-	}
-	cbReq := &cbapi.CallbackAfterSendGroupMsgReq{
-		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand),
-		GroupID:           msg.GroupID,
-	}
-
-	mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg))
-}
-
-func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string {
-	keyMsgData := apistruct.KeyMsgData{
-		SendID:  msg.SendID,
-		RecvID:  msg.RecvID,
-		GroupID: msg.GroupID,
-	}
-
-	return map[string]string{
-		webhook.Key: base64.StdEncoding.EncodeToString(stringutil.StructToJsonBytes(keyMsgData)),
-	}
-}
-
-func filterAfterMsg(msg *sdkws.MsgData, after *config.AfterConfig) bool {
-	return filterMsg(msg, after.AttentionIds, after.DeniedTypes)
-}
-
-func filterMsg(msg *sdkws.MsgData, attentionIds []string, deniedTypes []int32) bool {
-	// According to the attentionIds configuration, only some users are sent
-	if len(attentionIds) != 0 && !datautil.Contain(msg.RecvID, attentionIds...) {
-		return false
-	}
-
-	if defaultDeniedTypes(msg.ContentType) {
-		return false
-	}
-
-	if len(deniedTypes) != 0 && datautil.Contain(msg.ContentType, deniedTypes...) {
-		return false
-	}
-
-	return true
-}
-
-func defaultDeniedTypes(contentType int32) bool {
-	if contentType >= constant.NotificationBegin && contentType <= constant.NotificationEnd {
-		return true
-	}
-	if contentType == constant.Typing {
-		return true
-	}
-	return false
-}

internal/msgtransfer/init.go

@@ -134,7 +134,7 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 	if err != nil {
 		return err
 	}
-	historyMongoHandler := NewOnlineHistoryMongoConsumerHandler(msgTransferDatabase,config)
+	historyMongoHandler := NewOnlineHistoryMongoConsumerHandler(msgTransferDatabase)
 
 	msgTransfer := &MsgTransfer{
 		historyConsumer:      historyConsumer,

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -19,8 +19,6 @@ import (
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
-	"github.com/openimsdk/protocol/constant"
 	pbmsg "github.com/openimsdk/protocol/msg"
 	"github.com/openimsdk/tools/log"
 	"google.golang.org/protobuf/proto"
@@ -28,15 +26,11 @@ import (
 
 type OnlineHistoryMongoConsumerHandler struct {
 	msgTransferDatabase controller.MsgTransferDatabase
-	config              *Config
-	webhookClient       *webhook.Client
 }
 
-func NewOnlineHistoryMongoConsumerHandler(database controller.MsgTransferDatabase, config *Config) *OnlineHistoryMongoConsumerHandler {
+func NewOnlineHistoryMongoConsumerHandler(database controller.MsgTransferDatabase) *OnlineHistoryMongoConsumerHandler {
 	return &OnlineHistoryMongoConsumerHandler{
 		msgTransferDatabase: database,
-		config:              config,
-		webhookClient:       webhook.NewWebhookClient(config.WebhooksConfig.URL),
 	}
 }
 
@@ -59,16 +53,6 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(ctx context.Cont
 	} else {
 		prommetrics.MsgInsertMongoSuccessCounter.Inc()
 	}
-
-	for _, msgData := range msgFromMQ.MsgData {
-		switch msgData.SessionType {
-		case constant.SingleChatType:
-			mc.webhookAfterSendSingleMsg(ctx, &mc.config.WebhooksConfig.AfterSendSingleMsg, msgData)
-		case constant.ReadGroupChatType:
-			mc.webhookAfterSendGroupMsg(ctx, &mc.config.WebhooksConfig.AfterSendGroupMsg, msgData)
-		}
-	}
-
 	//var seqs []int64
 	//for _, msg := range msgFromMQ.MsgData {
 	//	seqs = append(seqs, msg.Seq)

internal/rpc/msg/send.go

@@ -86,8 +86,7 @@ func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *pbmsg.SendMsgReq,
 		go m.setConversationAtInfo(ctx, req.MsgData)
 	}
 
-	// m.webhookAfterSendGroupMsg(ctx, &m.config.WebhooksConfig.AfterSendGroupMsg, req)
-
+	m.webhookAfterSendGroupMsg(ctx, &m.config.WebhooksConfig.AfterSendGroupMsg, req)
 	prommetrics.GroupChatMsgProcessSuccessCounter.Inc()
 	resp = &pbmsg.SendMsgResp{}
 	resp.SendTime = req.MsgData.SendTime
@@ -193,8 +192,7 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
 			prommetrics.SingleChatMsgProcessFailedCounter.Inc()
 			return nil, err
 		}
-
-		// m.webhookAfterSendSingleMsg(ctx, &m.config.WebhooksConfig.AfterSendSingleMsg, req)
+		m.webhookAfterSendSingleMsg(ctx, &m.config.WebhooksConfig.AfterSendSingleMsg, req)
 		prommetrics.SingleChatMsgProcessSuccessCounter.Inc()
 		return &pbmsg.SendMsgResp{
 			ServerMsgID: req.MsgData.ServerMsgID,

internal/tools/cron/cron_task.go

@@ -58,11 +58,6 @@ func Start(ctx context.Context, conf *Config, client discovery.Conn, service grp
 		cm.Watch(ctx)
 	}
 
-	locker, err := NewEtcdLocker(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient())
-	if err != nil {
-		return err
-	}
-
 	srv := &cronServer{
 		ctx:                ctx,
 		config:             conf,
@@ -70,7 +65,6 @@ func Start(ctx context.Context, conf *Config, client discovery.Conn, service grp
 		msgClient:          msg.NewMsgClient(msgConn),
 		conversationClient: pbconversation.NewConversationClient(conversationConn),
 		thirdClient:        third.NewThirdClient(thirdConn),
-		locker:             locker,
 	}
 
 	if err := srv.registerClearS3(); err != nil {
@@ -87,8 +81,6 @@ func Start(ctx context.Context, conf *Config, client discovery.Conn, service grp
 	log.ZDebug(ctx, "cron task server is running")
 	<-ctx.Done()
 	log.ZDebug(ctx, "cron task server is shutting down")
-	srv.cron.Stop()
-
 	return nil
 }
 
@@ -99,7 +91,6 @@ type cronServer struct {
 	msgClient          msg.MsgClient
 	conversationClient pbconversation.ConversationClient
 	thirdClient        third.ThirdClient
-	locker             *EtcdLocker
 }
 
 func (c *cronServer) registerClearS3() error {
@@ -107,9 +98,7 @@ func (c *cronServer) registerClearS3() error {
 		log.ZInfo(c.ctx, "disable scheduled cleanup of s3", "fileExpireTime", c.config.CronTask.FileExpireTime, "deleteObjectType", c.config.CronTask.DeleteObjectType)
 		return nil
 	}
-	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, func() {
-		c.locker.ExecuteWithLock(c.ctx, "clearS3", c.clearS3)
-	})
+	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, c.clearS3)
 	return errs.WrapMsg(err, "failed to register clear s3 cron task")
 }
 
@@ -118,15 +107,11 @@ func (c *cronServer) registerDeleteMsg() error {
 		log.ZInfo(c.ctx, "disable scheduled cleanup of chat records", "retainChatRecords", c.config.CronTask.RetainChatRecords)
 		return nil
 	}
-	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, func() {
-		c.locker.ExecuteWithLock(c.ctx, "deleteMsg", c.deleteMsg)
-	})
+	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, c.deleteMsg)
 	return errs.WrapMsg(err, "failed to register delete msg cron task")
 }
 
 func (c *cronServer) registerClearUserMsg() error {
-	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, func() {
-		c.locker.ExecuteWithLock(c.ctx, "clearUserMsg", c.clearUserMsg)
-	})
+	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, c.clearUserMsg)
 	return errs.WrapMsg(err, "failed to register clear user msg cron task")
 }

internal/tools/cron/dist_look.go

@@ -1,89 +0,0 @@
-package cron
-
-import (
-	"context"
-	"fmt"
-	"os"
-	"time"
-
-	"github.com/openimsdk/tools/log"
-	clientv3 "go.etcd.io/etcd/client/v3"
-	"go.etcd.io/etcd/client/v3/concurrency"
-)
-
-const (
-	lockLeaseTTL = 300
-)
-
-type EtcdLocker struct {
-	client     *clientv3.Client
-	instanceID string
-}
-
-// NewEtcdLocker creates a new etcd distributed lock
-func NewEtcdLocker(client *clientv3.Client) (*EtcdLocker, error) {
-	hostname, _ := os.Hostname()
-	pid := os.Getpid()
-	instanceID := fmt.Sprintf("%s-pid-%d-%d", hostname, pid, time.Now().UnixNano())
-
-	locker := &EtcdLocker{
-		client:     client,
-		instanceID: instanceID,
-	}
-
-	return locker, nil
-}
-
-func (e *EtcdLocker) ExecuteWithLock(ctx context.Context, taskName string, task func()) {
-	session, err := concurrency.NewSession(e.client, concurrency.WithTTL(lockLeaseTTL))
-	if err != nil {
-		log.ZWarn(ctx, "Failed to create etcd session", err,
-			"taskName", taskName,
-			"instanceID", e.instanceID)
-		return
-	}
-	defer session.Close()
-
-	lockKey := fmt.Sprintf("openim/crontask/%s", taskName)
-	mutex := concurrency.NewMutex(session, lockKey)
-
-	ctxWithTimeout, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
-	defer cancel()
-
-	err = mutex.TryLock(ctxWithTimeout)
-	if err != nil {
-		if err == context.DeadlineExceeded {
-			log.ZDebug(ctx, "Task is being executed by another instance, skipping",
-				"taskName", taskName,
-				"instanceID", e.instanceID)
-		} else {
-			log.ZWarn(ctx, "Failed to acquire task lock", err,
-				"taskName", taskName,
-				"instanceID", e.instanceID)
-		}
-		return
-	}
-
-	defer func() {
-		if err := mutex.Unlock(ctx); err != nil {
-			log.ZWarn(ctx, "Failed to release task lock", err,
-				"taskName", taskName,
-				"instanceID", e.instanceID)
-		} else {
-			log.ZInfo(ctx, "Successfully released task lock",
-				"taskName", taskName,
-				"instanceID", e.instanceID)
-		}
-	}()
-
-	log.ZInfo(ctx, "Successfully acquired task lock, starting execution",
-		"taskName", taskName,
-		"instanceID", e.instanceID,
-		"sessionID", session.Lease())
-
-	task()
-
-	log.ZInfo(ctx, "Task execution completed",
-		"taskName", taskName,
-		"instanceID", e.instanceID)
-}

pkg/common/config/config.go

@@ -328,7 +328,7 @@ type Redis struct {
 	Username    string   `yaml:"username"`
 	Password    string   `yaml:"password"`
 	ClusterMode bool     `yaml:"clusterMode"`
-	DB          int      `yaml:"db"`
+	DB          int      `yaml:"storage"`
 	MaxRetry    int      `yaml:"maxRetry"`
 	PoolSize    int      `yaml:"poolSize"`
 }

start-config.yml

@@ -1,6 +1,6 @@
 serviceBinaries:
   openim-api: 1
-  openim-crontask: 4
+  openim-crontask: 1
   openim-rpc-user: 1
   openim-msggateway: 1
   openim-push: 8