blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-12-11 23:47:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' of https://github.com/openimsdk/open-im-server into fix/msg-geteway-panic

Browse Source
This commit is contained in:
mo3et 2025-11-05 10:47:45 +08:00
commit c50ebebf0f
16 changed files with 240 additions and 175 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

155
.github/workflows/publish-docker-image.yml vendored
View File

@ -4,42 +4,80 @@ on:
push: push:
branches: branches:
- release-* - release-*
# tags:
# - 'v*'
release: release:
types: [published] types: [published]
workflow_dispatch: workflow_dispatch:
inputs: inputs:
tag: tag:
description: "Tag version to be used for Docker image" description: "Tag version to be used for Docker image"
required: true required: true
default: "v3.8.0" default: "v3.8.3"
env:
GO_VERSION: "1.22"
IMAGE_NAME: "openim-server"
# IMAGE_NAME: ${{ github.event.repository.name }}
DOCKER_BUILDKIT: 1
jobs: jobs:
build-and-test: publish-docker-images:
runs-on: ubuntu-latest runs-on: ubuntu-latest
if: ${{ !(github.event_name == 'pull_request' && github.event.pull_request.merged == false) }}
steps: steps:
- uses: actions/checkout@v4 - name: Checkout main repository
uses: actions/checkout@v4
with: with:
path: main-repo path: main-repo
# - name: Set up QEMU - name: Set up QEMU
# uses: docker/setup-qemu-action@v3.3.0 uses: docker/setup-qemu-action@v3.3.0
- name: Set up Docker Buildx - name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3.8.0 id: buildx
uses: docker/setup-buildx-action@v3
- name: Build Docker image
id: build
uses: docker/build-push-action@v5
with: with:
context: ./main-repo driver-opts: network=host
load: true
tags: "openim/openim-server:local" - name: Extract metadata for Docker
cache-from: type=gha,scope=build id: meta
cache-to: type=gha,mode=max,scope=build uses: docker/metadata-action@v5.6.0
with:
images: |
${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}
ghcr.io/${{ github.repository_owner }}/${{ env.IMAGE_NAME }}
registry.cn-hangzhou.aliyuncs.com/openimsdk/${{ env.IMAGE_NAME }}
tags: |
type=ref,event=tag
type=schedule
type=ref,event=branch
type=ref,event=pr
type=semver,pattern={{version}}
type=semver,pattern=v{{version}}
type=semver,pattern={{major}}.{{minor}}
type=semver,pattern={{major}}
type=sha
- name: Install skopeo
run: |
sudo apt-get update && sudo apt-get install -y skopeo
- name: Build multi-arch images as OCI
run: |
mkdir -p /tmp/oci-image /tmp/docker-cache
# Build multi-architecture image and save in OCI format
docker buildx build \
--platform linux/amd64,linux/arm64 \
--output type=oci,dest=/tmp/oci-image/multi-arch.tar \
--cache-to type=local,dest=/tmp/docker-cache \
--cache-from type=gha \
./main-repo
# Use skopeo to convert the amd64 image from OCI format to Docker format and load it
skopeo copy --override-arch amd64 oci-archive:/tmp/oci-image/multi-arch.tar docker-daemon:${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}:local
# check image
docker image ls | grep openim
- name: Checkout compose repository - name: Checkout compose repository
uses: actions/checkout@v4 uses: actions/checkout@v4
@ -52,11 +90,11 @@ jobs:
run: | run: |
IP=$(hostname -I | awk '{print $1}') IP=$(hostname -I | awk '{print $1}')
echo "The IP Address is: $IP" echo "The IP Address is: $IP"
echo "::set-output name=ip::$IP" echo "ip=$IP" >> $GITHUB_OUTPUT
- name: Update .env to use the local image - name: Update .env to use the local image
run: | run: |
sed -i 's|OPENIM_SERVER_IMAGE=.*|OPENIM_SERVER_IMAGE=openim/openim-server:local|' ${{ github.workspace }}/compose-repo/.env sed -i 's|OPENIM_SERVER_IMAGE=.*|OPENIM_SERVER_IMAGE=${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}:local|' ${{ github.workspace }}/compose-repo/.env
sed -i 's|MINIO_EXTERNAL_ADDRESS=.*|MINIO_EXTERNAL_ADDRESS=http://${{ steps.get-ip.outputs.ip }}:10005|' ${{ github.workspace }}/compose-repo/.env sed -i 's|MINIO_EXTERNAL_ADDRESS=.*|MINIO_EXTERNAL_ADDRESS=http://${{ steps.get-ip.outputs.ip }}:10005|' ${{ github.workspace }}/compose-repo/.env
- name: Start services using Docker Compose - name: Start services using Docker Compose
@ -66,23 +104,34 @@ jobs:
docker compose ps docker compose ps
- name: Extract metadata for Docker (tags, labels) # - name: Check openim-server health
id: meta # run: |
uses: docker/metadata-action@v5.6.0 # timeout=300
with: # interval=30
images: | # elapsed=0
openim/openim-server # while [[ $elapsed -le $timeout ]]; do
ghcr.io/openimsdk/openim-server # if ! docker exec openim-server mage check; then
registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-server # echo "openim-server is not ready, waiting..."
tags: | # sleep $interval
type=ref,event=tag # elapsed=$(($elapsed + $interval))
type=schedule # else
type=ref,event=branch # echo "Health check successful"
# type=semver,pattern={{version}} # exit 0
type=semver,pattern=v{{version}} # fi
type=semver,pattern=release-{{raw}} # done
type=sha # echo "Health check failed after 5 minutes"
type=raw,value=${{ github.event.inputs.tag }} # exit 1
# - name: Check openim-chat health
# if: success()
# run: |
# if ! docker exec openim-chat mage check; then
# echo "openim-chat check failed"
# exit 1
# else
# echo "Health check successful"
# exit 0
# fi
- name: Log in to Docker Hub - name: Log in to Docker Hub
uses: docker/login-action@v3.3.0 uses: docker/login-action@v3.3.0
@ -104,22 +153,27 @@ jobs:
username: ${{ secrets.ALIREGISTRY_USERNAME }} username: ${{ secrets.ALIREGISTRY_USERNAME }}
password: ${{ secrets.ALIREGISTRY_TOKEN }} password: ${{ secrets.ALIREGISTRY_TOKEN }}
- name: Push Docker images - name: Push multi-architecture images
uses: docker/build-push-action@v5 if: success()
with: run: |
context: ./main-repo docker buildx build \
push: true --platform linux/amd64,linux/arm64 \
platforms: linux/amd64,linux/arm64 $(echo "${{ steps.meta.outputs.tags }}" | sed 's/,/ --tag /g' | sed 's/^/--tag /') \
tags: ${{ steps.meta.outputs.tags }} --cache-from type=local,src=/tmp/docker-cache \
labels: ${{ steps.meta.outputs.labels }} --push \
cache-from: type=gha,scope=build ./main-repo
cache-to: type=gha,mode=max,scope=build
- name: Verify multi-platform support - name: Verify multi-platform support
run: | run: |
images=("openim/openim-server" "ghcr.io/openimsdk/openim-server" "registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-server") images=(
"${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}"
"ghcr.io/${{ github.repository_owner }}/${{ env.IMAGE_NAME }}"
"registry.cn-hangzhou.aliyuncs.com/openimsdk/${{ env.IMAGE_NAME }}"
)
for image in "${images[@]}"; do for image in "${images[@]}"; do
for tag in $(echo "${{ steps.meta.outputs.tags }}" | tr ',' '\n'); do for tag in $(echo "${{ steps.meta.outputs.tags }}" | tr ',' '\n' | cut -d':' -f2); do
echo "Verifying multi-arch support for $image:$tag"
manifest=$(docker manifest inspect "$image:$tag" || echo "error") manifest=$(docker manifest inspect "$image:$tag" || echo "error")
if [[ "$manifest" == "error" ]]; then if [[ "$manifest" == "error" ]]; then
echo "Manifest not found for $image:$tag" echo "Manifest not found for $image:$tag"
@ -135,5 +189,6 @@ jobs:
echo "Multi-platform support check failed for $image:$tag - missing arm64" echo "Multi-platform support check failed for $image:$tag - missing arm64"
exit 1 exit 1
fi fi
echo "✅ $image:$tag supports both amd64 and arm64 architectures"
done done
done done

3
config/webhooks.yml
View File

@ -41,6 +41,9 @@ afterSendGroupMsg:
attentionIds: [] attentionIds: []
# See beforeSendSingleMsg comment. # See beforeSendSingleMsg comment.
deniedTypes: [] deniedTypes: []
afterMsgSaveDB:
enable: false
timeout: 5
afterUserOnline: afterUserOnline:
enable: false enable: false
timeout: 5 timeout: 5

2
internal/msggateway/client.go
View File

@ -70,6 +70,7 @@ type Client struct {
UserID string `json:"userID"` UserID string `json:"userID"`
IsBackground bool `json:"isBackground"` IsBackground bool `json:"isBackground"`
SDKType string `json:"sdkType"` SDKType string `json:"sdkType"`
SDKVersion string `json:"sdkVersion"`
Encoder Encoder Encoder Encoder
ctx *UserConnContext ctx *UserConnContext
longConnServer LongConnServer longConnServer LongConnServer
@ -97,6 +98,7 @@ func (c *Client) ResetClient(ctx *UserConnContext, conn LongConn, longConnServer
c.closedErr = nil c.closedErr = nil
c.token = ctx.GetToken() c.token = ctx.GetToken()
c.SDKType = ctx.GetSDKType() c.SDKType = ctx.GetSDKType()
c.SDKVersion = ctx.GetSDKVersion()
c.hbCtx, c.hbCancel = context.WithCancel(c.ctx) c.hbCtx, c.hbCancel = context.WithCancel(c.ctx)
c.subLock = new(sync.Mutex) c.subLock = new(sync.Mutex)
if c.subUserIDs != nil { if c.subUserIDs != nil {

1
internal/msggateway/constant.go
View File

@ -28,6 +28,7 @@ const (
BackgroundStatus = "isBackground" BackgroundStatus = "isBackground"
SendResponse = "isMsgResp" SendResponse = "isMsgResp"
SDKType = "sdkType" SDKType = "sdkType"
SDKVersion = "sdkVersion"
) )
const ( const (

7
internal/msggateway/context.go
View File

@ -15,12 +15,13 @@
package msggateway package msggateway
import ( import (
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"net/http" "net/http"
"net/url" "net/url"
"strconv" "strconv"
"time" "time"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/constant"
"github.com/openimsdk/tools/utils/encrypt" "github.com/openimsdk/tools/utils/encrypt"
"github.com/openimsdk/tools/utils/stringutil" "github.com/openimsdk/tools/utils/stringutil"
@ -140,6 +141,10 @@ func (c *UserConnContext) GetToken() string {
return c.Req.URL.Query().Get(Token) return c.Req.URL.Query().Get(Token)
} }
func (c *UserConnContext) GetSDKVersion() string {
return c.Req.URL.Query().Get(SDKVersion)
}
func (c *UserConnContext) GetCompression() bool { func (c *UserConnContext) GetCompression() bool {
compression, exists := c.Query(Compression) compression, exists := c.Query(Compression)
if exists && compression == GzipCompressionProtocol { if exists && compression == GzipCompressionProtocol {

4
internal/msggateway/ws_server.go
View File

@ -259,6 +259,10 @@ func (ws *WsServer) registerClient(client *Client) {
oldClients []*Client oldClients []*Client
) )
oldClients, userOK, clientOK = ws.clients.Get(client.UserID, client.PlatformID) oldClients, userOK, clientOK = ws.clients.Get(client.UserID, client.PlatformID)
log.ZInfo(client.ctx, "registerClient", "userID", client.UserID, "platformID", client.PlatformID,
"sdkVersion", client.SDKVersion)
if !userOK { if !userOK {
ws.clients.Set(client.UserID, client) ws.clients.Set(client.UserID, client)
log.ZDebug(client.ctx, "user not exist", "userID", client.UserID, "platformID", client.PlatformID) log.ZDebug(client.ctx, "user not exist", "userID", client.UserID, "platformID", client.PlatformID)

33
internal/msgtransfer/callback.go
View File

@ -51,37 +51,24 @@ func GetContent(msg *sdkws.MsgData) string {
} }
} }
func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) { func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterMsgSaveDB(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) {
if msg.ContentType == constant.Typing {
return
}
if !filterAfterMsg(msg, after) { if !filterAfterMsg(msg, after) {
return return
} }
cbReq := &cbapi.CallbackAfterSendSingleMsgReq{ cbReq := &cbapi.CallbackAfterMsgSaveDBReq{
CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand), CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterMsgSaveDBCommand),
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) { switch msg.SessionType {
return case constant.SingleChatType, constant.NotificationChatType:
cbReq.RecvID = msg.RecvID
case constant.ReadGroupChatType:
cbReq.GroupID = msg.GroupID
default:
} }
cbReq := &cbapi.CallbackAfterSendGroupMsgReq{ mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterMsgSaveDBResp{}, after, buildKeyMsgDataQuery(msg))
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 { func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string {

10
internal/msgtransfer/online_msg_to_mongo_handler.go
View File

@ -15,7 +15,6 @@
package msgtransfer package msgtransfer
import ( import (
"github.com/openimsdk/protocol/constant"
"github.com/openimsdk/tools/mq" "github.com/openimsdk/tools/mq"
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
@ -57,7 +56,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message)
log.ZDebug(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String()) log.ZDebug(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String())
err = mc.msgTransferDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq) err = mc.msgTransferDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq)
if err != nil { if err != nil {
log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID) log.ZError(ctx, "batch data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
prommetrics.MsgInsertMongoFailedCounter.Inc() prommetrics.MsgInsertMongoFailedCounter.Inc()
} else { } else {
prommetrics.MsgInsertMongoSuccessCounter.Inc() prommetrics.MsgInsertMongoSuccessCounter.Inc()
@ -65,12 +64,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message)
} }
for _, msgData := range msgFromMQ.MsgData { for _, msgData := range msgFromMQ.MsgData {
switch msgData.SessionType { mc.webhookAfterMsgSaveDB(ctx, &mc.config.WebhooksConfig.AfterMsgSaveDB, msgData)
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 //var seqs []int64

76
internal/rpc/msg/callback.go
View File

@ -16,8 +16,10 @@ package msg
import ( import (
"context" "context"
"encoding/base64"
"encoding/json" "encoding/json"
"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
"github.com/openimsdk/open-im-server/v3/pkg/common/webhook" "github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
"github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/errs"
@ -28,6 +30,7 @@ import (
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/utils/datautil" "github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/tools/utils/stringutil"
"google.golang.org/protobuf/proto" "google.golang.org/protobuf/proto"
) )
@ -87,19 +90,19 @@ func (m *msgServer) webhookBeforeSendSingleMsg(ctx context.Context, before *conf
} }
// Move to msgtransfer // Move to msgtransfer
// func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) { func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) {
// if msg.MsgData.ContentType == constant.Typing { if msg.MsgData.ContentType == constant.Typing {
// return return
// } }
// if !filterAfterMsg(msg, after) { if !filterAfterMsg(msg, after) {
// return return
// } }
// cbReq := &cbapi.CallbackAfterSendSingleMsgReq{ cbReq := &cbapi.CallbackAfterSendSingleMsgReq{
// CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand), CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand),
// RecvID: msg.MsgData.RecvID, RecvID: msg.MsgData.RecvID,
// } }
// m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData)) m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
// } }
func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq) error { func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq) error {
return webhook.WithCondition(ctx, before, func(ctx context.Context) error { return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
@ -121,21 +124,20 @@ func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *confi
}) })
} }
// Move to msgtransfer func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) {
// func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) { if msg.MsgData.ContentType == constant.Typing {
// if msg.MsgData.ContentType == constant.Typing { return
// return }
// } if !filterAfterMsg(msg, after) {
// if !filterAfterMsg(msg, after) { return
// return }
// } cbReq := &cbapi.CallbackAfterSendGroupMsgReq{
// cbReq := &cbapi.CallbackAfterSendGroupMsgReq{ CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand),
// CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand), GroupID: msg.MsgData.GroupID,
// GroupID: msg.MsgData.GroupID, }
// }
// m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData)) m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
// } }
func (m *msgServer) webhookBeforeMsgModify(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq, beforeMsgData **sdkws.MsgData) error { func (m *msgServer) webhookBeforeMsgModify(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq, beforeMsgData **sdkws.MsgData) error {
return webhook.WithCondition(ctx, before, func(ctx context.Context) error { return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
@ -204,14 +206,14 @@ func (m *msgServer) webhookAfterRevokeMsg(ctx context.Context, after *config.Aft
m.webhookClient.AsyncPost(ctx, callbackReq.GetCallbackCommand(), callbackReq, &cbapi.CallbackAfterRevokeMsgResp{}, after) m.webhookClient.AsyncPost(ctx, callbackReq.GetCallbackCommand(), callbackReq, &cbapi.CallbackAfterRevokeMsgResp{}, after)
} }
// func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string { func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string {
// keyMsgData := apistruct.KeyMsgData{ keyMsgData := apistruct.KeyMsgData{
// SendID: msg.SendID, SendID: msg.SendID,
// RecvID: msg.RecvID, RecvID: msg.RecvID,
// GroupID: msg.GroupID, GroupID: msg.GroupID,
// } }
// return map[string]string{ return map[string]string{
// webhook.Key: base64.StdEncoding.EncodeToString(stringutil.StructToJsonBytes(keyMsgData)), webhook.Key: base64.StdEncoding.EncodeToString(stringutil.StructToJsonBytes(keyMsgData)),
// } }
// } }

4
internal/rpc/msg/send.go
View File

@ -86,7 +86,7 @@ func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *pbmsg.SendMsgReq,
go m.setConversationAtInfo(ctx, req.MsgData) 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() prommetrics.GroupChatMsgProcessSuccessCounter.Inc()
resp = &pbmsg.SendMsgResp{} resp = &pbmsg.SendMsgResp{}
@ -194,7 +194,7 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
return nil, err return nil, err
} }
// m.webhookAfterSendSingleMsg(ctx, &m.config.WebhooksConfig.AfterSendSingleMsg, req) m.webhookAfterSendSingleMsg(ctx, &m.config.WebhooksConfig.AfterSendSingleMsg, req)
prommetrics.SingleChatMsgProcessSuccessCounter.Inc() prommetrics.SingleChatMsgProcessSuccessCounter.Inc()
return &pbmsg.SendMsgResp{ return &pbmsg.SendMsgResp{
ServerMsgID: req.MsgData.ServerMsgID, ServerMsgID: req.MsgData.ServerMsgID,

1
pkg/callbackstruct/constant.go
View File

@ -66,4 +66,5 @@ const (
CallbackAfterCreateSingleChatConversationsCommand = "callbackAfterCreateSingleChatConversationsCommand" CallbackAfterCreateSingleChatConversationsCommand = "callbackAfterCreateSingleChatConversationsCommand"
CallbackBeforeCreateGroupChatConversationsCommand = "callbackBeforeCreateGroupChatConversationsCommand" CallbackBeforeCreateGroupChatConversationsCommand = "callbackBeforeCreateGroupChatConversationsCommand"
CallbackAfterCreateGroupChatConversationsCommand = "callbackAfterCreateGroupChatConversationsCommand" CallbackAfterCreateGroupChatConversationsCommand = "callbackAfterCreateGroupChatConversationsCommand"
CallbackAfterMsgSaveDBCommand = "callbackAfterMsgSaveDBCommand"
) )

104
pkg/callbackstruct/conversation.go
View File

@ -2,42 +2,42 @@ package callbackstruct
type CallbackBeforeCreateSingleChatConversationsReq struct { type CallbackBeforeCreateSingleChatConversationsReq struct {
CallbackCommand `json:"callbackCommand"` CallbackCommand `json:"callbackCommand"`
OwnerUserID string `json:"owner_user_id"` OwnerUserID string `json:"ownerUserId"`
ConversationID string `json:"conversation_id"` ConversationID string `json:"conversationId"`
ConversationType int32 `json:"conversation_type"` ConversationType int32 `json:"conversationType"`
UserID string `json:"user_id"` UserID string `json:"userId"`
RecvMsgOpt int32 `json:"recv_msg_opt"` RecvMsgOpt int32 `json:"recvMsgOpt"`
IsPinned bool `json:"is_pinned"` IsPinned bool `json:"isPinned"`
IsPrivateChat bool `json:"is_private_chat"` IsPrivateChat bool `json:"isPrivateChat"`
BurnDuration int32 `json:"burn_duration"` BurnDuration int32 `json:"burnDuration"`
GroupAtType int32 `json:"group_at_type"` GroupAtType int32 `json:"groupAtType"`
AttachedInfo string `json:"attached_info"` AttachedInfo string `json:"attachedInfo"`
Ex string `json:"ex"` Ex string `json:"ex"`
} }
type CallbackBeforeCreateSingleChatConversationsResp struct { type CallbackBeforeCreateSingleChatConversationsResp struct {
CommonCallbackResp CommonCallbackResp
RecvMsgOpt *int32 `json:"recv_msg_opt"` RecvMsgOpt *int32 `json:"recvMsgOpt"`
IsPinned *bool `json:"is_pinned"` IsPinned *bool `json:"isPinned"`
IsPrivateChat *bool `json:"is_private_chat"` IsPrivateChat *bool `json:"isPrivateChat"`
BurnDuration *int32 `json:"burn_duration"` BurnDuration *int32 `json:"burnDuration"`
GroupAtType *int32 `json:"group_at_type"` GroupAtType *int32 `json:"groupAtType"`
AttachedInfo *string `json:"attached_info"` AttachedInfo *string `json:"attachedInfo"`
Ex *string `json:"ex"` Ex *string `json:"ex"`
} }
type CallbackAfterCreateSingleChatConversationsReq struct { type CallbackAfterCreateSingleChatConversationsReq struct {
CallbackCommand `json:"callbackCommand"` CallbackCommand `json:"callbackCommand"`
OwnerUserID string `json:"owner_user_id"` OwnerUserID string `json:"ownerUserId"`
ConversationID string `json:"conversation_id"` ConversationID string `json:"conversationId"`
ConversationType int32 `json:"conversation_type"` ConversationType int32 `json:"conversationType"`
UserID string `json:"user_id"` UserID string `json:"userId"`
RecvMsgOpt int32 `json:"recv_msg_opt"` RecvMsgOpt int32 `json:"recvMsgOpt"`
IsPinned bool `json:"is_pinned"` IsPinned bool `json:"isPinned"`
IsPrivateChat bool `json:"is_private_chat"` IsPrivateChat bool `json:"isPrivateChat"`
BurnDuration int32 `json:"burn_duration"` BurnDuration int32 `json:"burnDuration"`
GroupAtType int32 `json:"group_at_type"` GroupAtType int32 `json:"groupAtType"`
AttachedInfo string `json:"attached_info"` AttachedInfo string `json:"attachedInfo"`
Ex string `json:"ex"` Ex string `json:"ex"`
} }
@ -47,42 +47,42 @@ type CallbackAfterCreateSingleChatConversationsResp struct {
type CallbackBeforeCreateGroupChatConversationsReq struct { type CallbackBeforeCreateGroupChatConversationsReq struct {
CallbackCommand `json:"callbackCommand"` CallbackCommand `json:"callbackCommand"`
OwnerUserID string `json:"owner_user_id"` OwnerUserID string `json:"ownerUserId"`
ConversationID string `json:"conversation_id"` ConversationID string `json:"conversationId"`
ConversationType int32 `json:"conversation_type"` ConversationType int32 `json:"conversationType"`
GroupID string `json:"group_id"` GroupID string `json:"groupId"`
RecvMsgOpt int32 `json:"recv_msg_opt"` RecvMsgOpt int32 `json:"recvMsgOpt"`
IsPinned bool `json:"is_pinned"` IsPinned bool `json:"isPinned"`
IsPrivateChat bool `json:"is_private_chat"` IsPrivateChat bool `json:"isPrivateChat"`
BurnDuration int32 `json:"burn_duration"` BurnDuration int32 `json:"burnDuration"`
GroupAtType int32 `json:"group_at_type"` GroupAtType int32 `json:"groupAtType"`
AttachedInfo string `json:"attached_info"` AttachedInfo string `json:"attachedInfo"`
Ex string `json:"ex"` Ex string `json:"ex"`
} }
type CallbackBeforeCreateGroupChatConversationsResp struct { type CallbackBeforeCreateGroupChatConversationsResp struct {
CommonCallbackResp CommonCallbackResp
RecvMsgOpt *int32 `json:"recv_msg_opt"` RecvMsgOpt *int32 `json:"recvMsgOpt"`
IsPinned *bool `json:"is_pinned"` IsPinned *bool `json:"isPinned"`
IsPrivateChat *bool `json:"is_private_chat"` IsPrivateChat *bool `json:"isPrivateChat"`
BurnDuration *int32 `json:"burn_duration"` BurnDuration *int32 `json:"burnDuration"`
GroupAtType *int32 `json:"group_at_type"` GroupAtType *int32 `json:"groupAtType"`
AttachedInfo *string `json:"attached_info"` AttachedInfo *string `json:"attachedInfo"`
Ex *string `json:"ex"` Ex *string `json:"ex"`
} }
type CallbackAfterCreateGroupChatConversationsReq struct { type CallbackAfterCreateGroupChatConversationsReq struct {
CallbackCommand `json:"callbackCommand"` CallbackCommand `json:"callbackCommand"`
OwnerUserID string `json:"owner_user_id"` OwnerUserID string `json:"ownerUserId"`
ConversationID string `json:"conversation_id"` ConversationID string `json:"conversationId"`
ConversationType int32 `json:"conversation_type"` ConversationType int32 `json:"conversationType"`
GroupID string `json:"group_id"` GroupID string `json:"groupId"`
RecvMsgOpt int32 `json:"recv_msg_opt"` RecvMsgOpt int32 `json:"recvMsgOpt"`
IsPinned bool `json:"is_pinned"` IsPinned bool `json:"isPinned"`
IsPrivateChat bool `json:"is_private_chat"` IsPrivateChat bool `json:"isPrivateChat"`
BurnDuration int32 `json:"burn_duration"` BurnDuration int32 `json:"burnDuration"`
GroupAtType int32 `json:"group_at_type"` GroupAtType int32 `json:"groupAtType"`
AttachedInfo string `json:"attached_info"` AttachedInfo string `json:"attachedInfo"`
Ex string `json:"ex"` Ex string `json:"ex"`
} }

10
pkg/callbackstruct/message.go
View File

@ -103,3 +103,13 @@ type CallbackSingleMsgReadReq struct {
type CallbackSingleMsgReadResp struct { type CallbackSingleMsgReadResp struct {
CommonCallbackResp CommonCallbackResp
} }
type CallbackAfterMsgSaveDBReq struct {
CommonCallbackReq
RecvID string `json:"recvID"`
GroupID string `json:"groupID"`
}
type CallbackAfterMsgSaveDBResp struct {
CommonCallbackResp
}

1
pkg/common/config/config.go
View File

@ -436,6 +436,7 @@ type Webhooks struct {
BeforeSendGroupMsg BeforeConfig `yaml:"beforeSendGroupMsg"` BeforeSendGroupMsg BeforeConfig `yaml:"beforeSendGroupMsg"`
BeforeMsgModify BeforeConfig `yaml:"beforeMsgModify"` BeforeMsgModify BeforeConfig `yaml:"beforeMsgModify"`
AfterSendGroupMsg AfterConfig `yaml:"afterSendGroupMsg"` AfterSendGroupMsg AfterConfig `yaml:"afterSendGroupMsg"`
AfterMsgSaveDB AfterConfig `yaml:"afterMsgSaveDB"`
AfterUserOnline AfterConfig `yaml:"afterUserOnline"` AfterUserOnline AfterConfig `yaml:"afterUserOnline"`
AfterUserOffline AfterConfig `yaml:"afterUserOffline"` AfterUserOffline AfterConfig `yaml:"afterUserOffline"`
AfterUserKickOff AfterConfig `yaml:"afterUserKickOff"` AfterUserKickOff AfterConfig `yaml:"afterUserKickOff"`

2
pkg/common/storage/database/mgo/user.go
View File

@ -73,7 +73,7 @@ func (u *UserMgo) TakeNotification(ctx context.Context, level int64) (user []*mo
} }
func (u *UserMgo) TakeGTEAppManagerLevel(ctx context.Context, level int64) (user []*model.User, err error) { func (u *UserMgo) TakeGTEAppManagerLevel(ctx context.Context, level int64) (user []*model.User, err error) {
return mongoutil.Find[*model.User](ctx, u.coll, bson.M{"app_manager_level": bson.M{"$gte": level}}) return mongoutil.Find[*model.User](ctx, u.coll, bson.M{"app_manger_level": bson.M{"$gte": level}})
} }
func (u *UserMgo) TakeByNickname(ctx context.Context, nickname string) (user []*model.User, err error) { func (u *UserMgo) TakeByNickname(ctx context.Context, nickname string) (user []*model.User, err error) {

2
pkg/msgprocessor/conversation.go
View File

@ -109,7 +109,7 @@ func GetConversationIDBySessionType(sessionType int, ids ...string) string {
case constant.ReadGroupChatType: case constant.ReadGroupChatType:
return "sg_" + ids[0] // super group chat return "sg_" + ids[0] // super group chat
case constant.NotificationChatType: case constant.NotificationChatType:
return "sn_" + ids[0] // server notification chat return "sn_" + strings.Join(ids, "_") // server notification chat
} }
return "" return ""
} }