Merge 8cd57d68535f83c6495423e777f6a30c88cbe0f1 into 75367545ea330bdb67a6e4242cb31e5652ce44dc

* feat: support distributed lock in crontask.

* remove space.

* remove comment.

* remove log.

* Update logic.

* Update contents.

internal/msgtransfer/callback.go

@@ -0,0 +1,124 @@
+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)
+	historyMongoHandler := NewOnlineHistoryMongoConsumerHandler(msgTransferDatabase,config)
 
 	msgTransfer := &MsgTransfer{
 		historyConsumer:      historyConsumer,

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -19,6 +19,8 @@ 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"
@@ -26,11 +28,15 @@ import (
 
 type OnlineHistoryMongoConsumerHandler struct {
 	msgTransferDatabase controller.MsgTransferDatabase
+	config              *Config
+	webhookClient       *webhook.Client
 }
 
-func NewOnlineHistoryMongoConsumerHandler(database controller.MsgTransferDatabase) *OnlineHistoryMongoConsumerHandler {
+func NewOnlineHistoryMongoConsumerHandler(database controller.MsgTransferDatabase, config *Config) *OnlineHistoryMongoConsumerHandler {
 	return &OnlineHistoryMongoConsumerHandler{
 		msgTransferDatabase: database,
+		config:              config,
+		webhookClient:       webhook.NewWebhookClient(config.WebhooksConfig.URL),
 	}
 }
 
@@ -53,6 +59,16 @@ 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,7 +86,8 @@ 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
@@ -192,7 +193,8 @@ 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,6 +58,11 @@ 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,
@@ -65,6 +70,7 @@ 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 {
@@ -81,6 +87,8 @@ 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
 }
 
@@ -91,6 +99,7 @@ type cronServer struct {
 	msgClient          msg.MsgClient
 	conversationClient pbconversation.ConversationClient
 	thirdClient        third.ThirdClient
+	locker             *EtcdLocker
 }
 
 func (c *cronServer) registerClearS3() error {
@@ -98,7 +107,9 @@ 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, c.clearS3)
+	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, func() {
+		c.locker.ExecuteWithLock(c.ctx, "clearS3", c.clearS3)
+	})
 	return errs.WrapMsg(err, "failed to register clear s3 cron task")
 }
 
@@ -107,11 +118,15 @@ 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, c.deleteMsg)
+	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, func() {
+		c.locker.ExecuteWithLock(c.ctx, "deleteMsg", 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, c.clearUserMsg)
+	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, func() {
+		c.locker.ExecuteWithLock(c.ctx, "clearUserMsg", c.clearUserMsg)
+	})
 	return errs.WrapMsg(err, "failed to register clear user msg cron task")
 }

internal/tools/cron/dist_look.go

@@ -0,0 +1,89 @@
+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:"storage"`
+	DB          int      `yaml:"db"`
 	MaxRetry    int      `yaml:"maxRetry"`
 	PoolSize    int      `yaml:"poolSize"`
 }

pkg/common/storage/cache/redis/batch.go

@@ -76,6 +76,27 @@ func batchGetCache2[K comparable, V any](ctx context.Context, rcClient *rocksCac
 	if err != nil {
 		return nil, err
 	}
+	values, err := fn(ctx, ids)
+	if err != nil {
+		log.ZError(ctx, "batchGetCache query database failed", err, "ids", ids)
+		return nil, err
+	}
+	idToValue := make(map[K]*V)
+	for _, value := range values {
+		idToValue[vId(value)] = value
+	}
+	getSlotValues := func(slotIds []K) []*V {
+		if len(slotIds) == 0 {
+			return nil
+		}
+		slotValues := make([]*V, 0, len(slotIds))
+		for _, id := range slotIds {
+			if value, ok := idToValue[id]; ok {
+				slotValues = append(slotValues, value)
+			}
+		}
+		return slotValues
+	}
 	result := make([]*V, 0, len(findKeys))
 	for _, keys := range slotKeys {
 		indexCache, err := rcClient.GetClient().FetchBatch2(ctx, keys, expire, func(idx []int) (map[int]string, error) {
@@ -86,16 +107,12 @@ func batchGetCache2[K comparable, V any](ctx context.Context, rcClient *rocksCac
 				idIndex[id] = index
 				queryIds = append(queryIds, id)
 			}
-			values, err := fn(ctx, queryIds)
-			if err != nil {
-				log.ZError(ctx, "batchGetCache query database failed", err, "keys", keys, "queryIds", queryIds)
-				return nil, err
-			}
-			if len(values) == 0 {
+			slotValues := getSlotValues(queryIds)
+			if len(slotValues) == 0 {
 				return map[int]string{}, nil
 			}
 			cacheIndex := make(map[int]string)
-			for _, value := range values {
+			for _, value := range slotValues {
 				id := vId(value)
 				index, ok := idIndex[id]
 				if !ok {

start-config.yml

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