feat: improve group member events and push delivery

This commit is contained in:
richjohnrambo
2026-09-18 01:00:41 +08:00
parent 0e2f4e73d1
commit c6e7f6b923
13 changed files with 521 additions and 19 deletions
+8 -3
View File
@@ -438,7 +438,9 @@ X-MeetingIM-APIKey: <预置的普通 API Key>
"event": "group.member.joined", "event": "group.member.joined",
"payload": { "payload": {
"topic": "grp_xxx", "topic": "grp_xxx",
"user_id": "usr_xxx" "user_id": "usr_xxx",
"member_sync": "single",
"visibility": "all_members"
}, },
"ts": "2026-08-25T10:00:00Z" "ts": "2026-08-25T10:00:00Z"
} }
@@ -456,7 +458,9 @@ X-MeetingIM-APIKey: <预置的普通 API Key>
| 消息撤回 | Topic 在线成员 | `{"info":{"topic":"p2p_or_grp","from":"usr_actor","what":"im","event":"message.recalled","payload":{"topic":"p2p_or_grp","seq":12,"version":2}}}` | 按 `topic+seq+version` 重新拉取或本地标记已撤回 | | 消息撤回 | Topic 在线成员 | `{"info":{"topic":"p2p_or_grp","from":"usr_actor","what":"im","event":"message.recalled","payload":{"topic":"p2p_or_grp","seq":12,"version":2}}}` | 按 `topic+seq+version` 重新拉取或本地标记已撤回 |
| 消息编辑 | Topic 在线成员 | `{"info":{"topic":"p2p_or_grp","from":"usr_actor","what":"im","event":"message.edited","payload":{"topic":"p2p_or_grp","seq":12,"version":2}}}` | 按 `topic+seq+version` 重新拉取消息详情或修订记录 | | 消息编辑 | Topic 在线成员 | `{"info":{"topic":"p2p_or_grp","from":"usr_actor","what":"im","event":"message.edited","payload":{"topic":"p2p_or_grp","seq":12,"version":2}}}` | 按 `topic+seq+version` 重新拉取消息详情或修订记录 |
| 单聊消息变更脏标记 | P2P 双方在线会话 | `{"info":{"topic":"usr_target","from":"usr_actor","what":"im","event":"message.mutation","payload":{"topic":"p2p...","seq":12,"version":2,"event":"message.recalled"}}}` | 当前未订阅该 P2P 也要标记会话脏,并调用 `message-mutations` 做补偿同步 | | 单聊消息变更脏标记 | P2P 双方在线会话 | `{"info":{"topic":"usr_target","from":"usr_actor","what":"im","event":"message.mutation","payload":{"topic":"p2p...","seq":12,"version":2,"event":"message.recalled"}}}` | 当前未订阅该 P2P 也要标记会话脏,并调用 `message-mutations` 做补偿同步 |
| 群业务变化 | 群 Topic 在线成员 | `{"info":{"topic":"grp...","from":"usr_actor","what":"group","event":"group.member.joined","payload":{"topic":"grp...","user_id":"usr_xxx"}}}` | 按事件刷新群详情、成员列表、入群申请、公告、禁言或角色权限;除操作人外当前会写系统通知 | | 成员加入 | 群 Topic 在线成员 | `{"info":{"topic":"grp...","from":"usr_actor","what":"group","event":"group.member.joined","payload":{"user_id":"usr_xxx","member_sync":"single","visibility":"all_members"}}}` | 全体成员可见;调用 `GET /v1/im/groups/{topic}/members/{user_id}` 刷新单个成员。加入事件同时写入群消息,离线后可按 Topic seq 补拉 |
| 成员退出/移除、角色或禁言变化 | 群主、管理员和被操作人 | `{"info":{"topic":"grp...","from":"usr_actor","what":"group","event":"group.member.left","payload":{"user_id":"usr_xxx","member_sync":"remove","visibility":"admins_and_affected"}}}` | 普通成员不可见;`single` 拉取单成员,`remove` 本地删除,`full` 重新分页拉取全部成员 |
| 其他群业务变化 | 按具体事件授权范围 | `info.what="group"` | 刷新群详情、入群申请、公告或治理权限 |
| 会议变化 | 会议参会人 | `{"info":{"topic":"usr_target","from":"usr_actor","what":"im","event":"meeting.updated","payload":{"meeting_id":"mtg_1","version":2,"state":"scheduled"}}}` | 刷新会议列表/详情;除操作者外当前会写系统通知 | | 会议变化 | 会议参会人 | `{"info":{"topic":"usr_target","from":"usr_actor","what":"im","event":"meeting.updated","payload":{"meeting_id":"mtg_1","version":2,"state":"scheduled"}}}` | 刷新会议列表/详情;除操作者外当前会写系统通知 |
| 安全风险确认 | 操作人本人 | `{"info":{"topic":"usr_actor","from":"usr_actor","what":"im","event":"security.risk.continue","payload":{"risk_challenge_id":"risk_1","decision":"continue"}}}` | 刷新安全验证页面;当前会写系统通知 | | 安全风险确认 | 操作人本人 | `{"info":{"topic":"usr_actor","from":"usr_actor","what":"im","event":"security.risk.continue","payload":{"risk_challenge_id":"risk_1","decision":"continue"}}}` | 刷新安全验证页面;当前会写系统通知 |
@@ -2173,7 +2177,8 @@ GET /v1/im/topics/{topic}/message-mutations?since_version={localMutationVersion}
| `PATCH /v1/im/groups/{topic}/member-settings` | `{"pinned":true,"mute_notifications":false,"version":1}` | `{"topic":"grp...","pinned":true,"mute_notifications":false,"version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `PATCH /v1/im/groups/{topic}/member-settings` | `{"pinned":true,"mute_notifications":false,"version":1}` | `{"topic":"grp...","pinned":true,"mute_notifications":false,"version":2,"updated_at":"2026-08-20T09:00:00Z"}` |
| `PUT /v1/im/groups/{topic}/members/me/profile` | `{"nickname":"我在群里的昵称","version":1}` | `{"nickname":"我在群里的昵称","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `PUT /v1/im/groups/{topic}/members/me/profile` | `{"nickname":"我在群里的昵称","version":1}` | `{"nickname":"我在群里的昵称","version":2,"updated_at":"2026-08-20T09:00:00Z"}` |
| `POST /v1/im/groups/{topic}/leave` | `{}` | `{"topic":"grp...","status":"left"}` | | `POST /v1/im/groups/{topic}/leave` | `{}` | `{"topic":"grp...","status":"left"}` |
| `GET /v1/im/groups/{topic}/members?query=&cursor=&mentionable=true` | Query only | `{"topic":"grp...","items":[{"user_id":"usr...","display_name":"张三","group_nickname":"小张","avatar_file_id":"","role":"member","mentionable":true,"acs":{"want":"JRWP","given":"JRWP","mode":"JRWP"}}],"next_cursor":"","permissions":{"can_mention_all":true}}` | | `GET /v1/im/groups/{topic}/members?query=&cursor=&mentionable=true` | Query only | `{"topic":"grp...","items":[{"user_id":"usr...","display_name":"张三","group_nickname":"小张","avatar_file_id":"","role":"member","member_status":"active","version":1,"joined_at":"2026-08-20T09:00:00Z","updated_at":"2026-08-20T09:00:00Z","mentionable":true,"acs":{"want":"JRWP","given":"JRWP","mode":"JRWP"}}],"next_cursor":"","sync_mode":"full","permissions":{"can_mention_all":true}}` |
| `GET /v1/im/groups/{topic}/members/{user_id}` | Path topic/user | `{"topic":"grp...","user_id":"usr...","display_name":"张三","group_nickname":"小张","role":"member","member_status":"active","version":1,"joined_at":"2026-08-20T09:00:00Z","updated_at":"2026-08-20T09:00:00Z","sync_mode":"single","acs":{"want":"JRWP","given":"JRWP","mode":"JRWP"}}` |
| `GET /v1/im/groups/{topic}/members/{user_id}/profile` | Path topic/user;`topic` 必须是 `grp...`,不能传 `usr...` | `{"topic":"grp...","user_id":"usr...","display_name":"张三","group_nickname":"小张","role":"member","relationship":"friend","capabilities":{"can_send":true,"can_call":true}}` | | `GET /v1/im/groups/{topic}/members/{user_id}/profile` | Path topic/user;`topic` 必须是 `grp...`,不能传 `usr...` | `{"topic":"grp...","user_id":"usr...","display_name":"张三","group_nickname":"小张","role":"member","relationship":"friend","capabilities":{"can_send":true,"can_call":true}}` |
| `POST /v1/im/groups/{topic}/qrcodes` | `{"expires_in_seconds":86400}` | `{"qr_token":"token...","qr_payload":"im://qr?token=...&purpose=join_group","expires_at":"2026-08-21T09:00:00Z","group":{"topic":"grp...","name":"项目群","member_count":2},"group_snapshot":{"topic":"grp...","name":"项目群","member_count":2}}` | | `POST /v1/im/groups/{topic}/qrcodes` | `{"expires_in_seconds":86400}` | `{"qr_token":"token...","qr_payload":"im://qr?token=...&purpose=join_group","expires_at":"2026-08-21T09:00:00Z","group":{"topic":"grp...","name":"项目群","member_count":2},"group_snapshot":{"topic":"grp...","name":"项目群","member_count":2}}` |
| `POST /v1/im/groups/{topic}/invitations` | `{"user_ids":["usr..."],"message":"进群一起讨论"}` | `{"topic":"grp...","items":[{"user_id":"usr...","status":"joined","invitation_id":"ginv_1"}]}` | | `POST /v1/im/groups/{topic}/invitations` | `{"user_ids":["usr..."],"message":"进群一起讨论"}` | `{"topic":"grp...","items":[{"user_id":"usr...","status":"joined","invitation_id":"ginv_1"}]}` |
+10 -8
View File
@@ -121,19 +121,21 @@
| --- | --- | --- | --- | --- | | --- | --- | --- | --- | --- |
| 创建群 | 群成员 | `info.what="group"`, `event="group.created"` | `topic`、群基础信息 | 除操作人外写入 | | 创建群 | 群成员 | `info.what="group"`, `event="group.created"` | `topic`、群基础信息 | 除操作人外写入 |
| 更新群设置 | 群成员 | `event="group.settings.updated"` | `topic` | 除操作人外写入 | | 更新群设置 | 群成员 | `event="group.settings.updated"` | `topic` | 除操作人外写入 |
| 成员资料/权限更新 | 群成员 | `event="group.member.updated"` | `user_id` | 除操作人外写入 | | 成员资料/权限更新 | 群主、管理员、本人 | `event="group.member.updated"` | `user_id`、`member_sync=single` | 不写群系统消息 |
| 成员加入 | 群成员 | `event="group.member.joined"` | `user_id`、`topic` | 除操作人外写入 | | 成员加入 | 群成员 | `event="group.member.joined"` | `user_id`、`topic`、`member_sync=single` | 写入群消息并参与个人未读与离线 Push |
| 入群申请创建 | 群成员/管理员可见链路 | `event="group.join_request.created"` | `request_id`、`applicant_user_id`、`topic` | 除操作人外写入 | | 入群申请创建 | 群主、管理员、申请人 | `event="group.join_request.created"` | `request_id`、`applicant_user_id`、`topic` | 普通成员不可见 |
| 入群申请处理 | 群成员 | `event="group.join_request.accepted/rejected"` | `request_id`、`status`、`topic` | 除操作人外写入 | | 入群申请处理 | 群主、管理员、申请人 | `event="group.join_request.accepted/rejected"` | `request_id`、`status`、`topic` | 普通成员不可见;接受时另发全群加入消息 |
| 群公告更新 | 群成员 | `event="group.announcement.updated"` | `topic`、公告信息 | 除操作人外写入 | | 群公告更新 | 群成员 | `event="group.announcement.updated"` | `topic`、公告信息 | 除操作人外写入 |
| 群禁言更新 | 群成员 | `event="group.mute.updated"` | `topic`、禁言信息 | 除操作人外写入 | | 群禁言更新 | 群主、管理员、被操作人 | `event="group.mute.updated"` | `topic`、禁言信息、`member_sync` | 普通成员不可见 |
| 成员离群 | 群成员 | `event="group.member.left"` | `user_id` | 除操作人外写入 | | 成员离群/移除 | 群主、管理员、离群成员 | `event="group.member.left"` | `user_id`、`member_sync=remove` | 普通成员不可见 |
| 群角色更新 | 群成员 | `event="group.role.updated"` | `user_id`、角色信息 | 除操作人外写入 | | 群角色更新 | 群主、管理员、被操作人 | `event="group.role.updated"` | `user_id`、角色信息、`member_sync` | 普通成员不可见 |
| 群主转让 | 群成员 | `event="group.owner.transferred"` | `previous_owner`、`new_owner` | 除操作人外写入 | | 群主转让 | 群成员 | `event="group.owner.transferred"` | `previous_owner`、`new_owner` | 除操作人外写入 |
前端建议动作: 前端建议动作:
- 群成员、角色、禁言、群主变化:刷新群详情、成员列表、当前用户群权限。 - 首次进入、断线重连、本地无快照或收到 `member_sync=full`:分页拉取完整群成员。
- 收到 `member_sync=single`:调用 `GET /v1/im/groups/{topic}/members/{user_id}` 只刷新一个成员。
- 收到 `member_sync=remove`:从本地成员快照删除 `user_id`;若本地状态无法确认,再全量分页同步。
- 入群申请变化:管理员刷新入群申请列表和群 summary 待处理数。 - 入群申请变化:管理员刷新入群申请列表和群 summary 待处理数。
- 群公告变化:刷新公告区。 - 群公告变化:刷新公告区。
- 成员加入/离开:刷新群成员数、成员列表和会话系统提示。 - 成员加入/离开:刷新群成员数、成员列表和会话系统提示。
+6 -3
View File
@@ -2950,7 +2950,7 @@ func (a *adapter) SubsForTopic(topic string, keepDeleted bool, opts *t.QueryOpt)
// Filter out deleted rows. // Filter out deleted rows.
q += " AND deletedat IS NULL" q += " AND deletedat IS NULL"
} }
limit := a.maxResults limit := 0
if opts != nil { if opts != nil {
// Ignore IfModifiedSince - we must return all entries // Ignore IfModifiedSince - we must return all entries
// Those unmodified will be stripped of Public & Private. // Those unmodified will be stripped of Public & Private.
@@ -2959,13 +2959,16 @@ func (a *adapter) SubsForTopic(topic string, keepDeleted bool, opts *t.QueryOpt)
q += " AND userid=?" q += " AND userid=?"
args = append(args, store.DecodeUid(opts.User)) args = append(args, store.DecodeUid(opts.User))
} }
limit = a.maxResults
if opts.Limit > 0 && opts.Limit < limit { if opts.Limit > 0 && opts.Limit < limit {
limit = opts.Limit limit = opts.Limit
} }
} }
q += " LIMIT ?" if limit > 0 {
args = append(args, limit) q += " LIMIT ?"
args = append(args, limit)
}
ctx, cancel := a.getContext() ctx, cancel := a.getContext()
if cancel != nil { if cancel != nil {
+6 -3
View File
@@ -1779,8 +1779,9 @@ func (a *adapter) TenantSubsForTopic(id t.TenantID, topic string, keepDeleted bo
if !keepDeleted { if !keepDeleted {
q += " AND deletedat IS NULL" q += " AND deletedat IS NULL"
} }
limit := a.maxResults limit := 0
if opts != nil { if opts != nil {
limit = a.maxResults
if !opts.User.IsZero() { if !opts.User.IsZero() {
q += " AND userid=?" q += " AND userid=?"
args = append(args, decodeUID(opts.User)) args = append(args, decodeUID(opts.User))
@@ -1789,8 +1790,10 @@ func (a *adapter) TenantSubsForTopic(id t.TenantID, topic string, keepDeleted bo
limit = opts.Limit limit = opts.Limit
} }
} }
q += " LIMIT ?" if limit > 0 {
args = append(args, limit) q += " LIMIT ?"
args = append(args, limit)
}
ctx, cancel := a.getContext() ctx, cancel := a.getContext()
if cancel != nil { if cancel != nil {
defer cancel() defer cancel()
+130
View File
@@ -2,10 +2,12 @@ package main
import ( import (
"encoding/json" "encoding/json"
"fmt"
"strings" "strings"
"time" "time"
"github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/logs"
"github.com/tinode/chat/server/store"
"github.com/tinode/chat/server/store/types" "github.com/tinode/chat/server/store/types"
) )
@@ -15,6 +17,7 @@ func publishIMGroupRealtimeEvent(tenantID types.TenantID, topic string, actor ty
if tenantID.IsZero() || topic == "" || eventType == "" { if tenantID.IsZero() || topic == "" || eventType == "" {
return return
} }
payload = enrichIMGroupMemberEventPayload(eventType, payload)
encoded, err := json.Marshal(payload) encoded, err := json.Marshal(payload)
if err != nil { if err != nil {
logs.Warn.Printf("group realtime: encode event topic=%s event=%s: %v", topic, eventType, err) logs.Warn.Printf("group realtime: encode event topic=%s event=%s: %v", topic, eventType, err)
@@ -24,6 +27,10 @@ func publishIMGroupRealtimeEvent(tenantID types.TenantID, topic string, actor ty
if !actor.IsZero() { if !actor.IsZero() {
from = actor.UserId() from = actor.UserId()
} }
if isAdminOnlyIMGroupEvent(eventType) {
publishIMGroupEventToAdmins(tenantID, topic, actor, eventType, encoded, payload)
return
}
message := &ServerComMessage{ message := &ServerComMessage{
TenantID: tenantID, TenantID: tenantID,
RcptTo: topic, RcptTo: topic,
@@ -47,6 +54,129 @@ func publishIMGroupRealtimeEvent(tenantID types.TenantID, topic string, actor ty
publishRealtimeTopicMessage(topic, message) publishRealtimeTopicMessage(topic, message)
} }
func isAdminOnlyIMGroupEvent(eventType string) bool {
switch eventType {
case "group.member.updated", "group.member.left", "group.role.updated", "group.mute.updated",
"group.invitations.updated", "group.join_request.created", "group.join_request.accepted",
"group.join_request.rejected":
return true
default:
return false
}
}
func publishIMGroupEventToAdmins(tenantID types.TenantID, topic string, actor types.Uid, eventType string,
encoded json.RawMessage, payload any) {
recipients := map[types.Uid]struct{}{}
if !actor.IsZero() {
recipients[actor] = struct{}{}
}
collectIMGroupEventUserIDs(payload, recipients)
for _, role := range []string{types.GroupMemberRoleOwner, types.GroupMemberRoleAdmin} {
if store.GroupBusiness == nil {
break
}
cursor := ""
for {
members, next, err := store.GroupBusiness.ListMembers(tenantID, topic, types.GroupMemberListOptions{
Role: role, Cursor: cursor, Limit: 100,
})
if err != nil {
logs.Warn.Printf("group realtime: list %s recipients topic=%s event=%s: %v", role, topic, eventType, err)
break
}
for _, member := range members {
if !member.UserID.IsZero() {
recipients[member.UserID] = struct{}{}
}
}
if next == "" || next == cursor {
break
}
cursor = next
}
}
for uid := range recipients {
message := &ServerComMessage{
TenantID: tenantID, RcptTo: uid.UserId(), AsUser: actor.UserId(),
Timestamp: time.Now().UTC().Round(time.Millisecond),
Info: &MsgServerInfo{Topic: topic, From: actor.UserId(), What: "group", Event: eventType, Payload: encoded},
}
if globals.sessionStore != nil {
globals.sessionStore.DeliverToUser(tenantID, uid, "", message)
}
publishRealtimeUserMessage(uid, message)
}
}
func collectIMGroupEventUserIDs(value any, out map[types.Uid]struct{}) {
switch typed := value.(type) {
case map[string]any:
if ok, exists := typed["ok"].(bool); exists && !ok {
return
}
for key, item := range typed {
switch key {
case "user_id", "target_id", "target_user_id", "applicant_id", "applicant_user_id", "owner_id":
if uid := imGroupEventUID(item); !uid.IsZero() {
out[uid] = struct{}{}
}
}
collectIMGroupEventUserIDs(item, out)
}
case []map[string]any:
for _, item := range typed {
collectIMGroupEventUserIDs(item, out)
}
case []any:
for _, item := range typed {
collectIMGroupEventUserIDs(item, out)
}
}
}
func imGroupEventUID(value any) types.Uid {
if uid := types.ParseUserId(strings.TrimSpace(fmt.Sprint(value))); !uid.IsZero() {
return uid
}
switch typed := value.(type) {
case int64:
return store.EncodeUid(typed)
case int:
return store.EncodeUid(int64(typed))
case float64:
return store.EncodeUid(int64(typed))
}
return types.ZeroUid
}
func enrichIMGroupMemberEventPayload(eventType string, payload any) any {
data, ok := payload.(map[string]any)
if !ok {
return payload
}
copyPayload := make(map[string]any, len(data)+2)
for key, value := range data {
copyPayload[key] = value
}
switch eventType {
case "group.member.left":
copyPayload["member_sync"] = "remove"
case "group.member.joined", "group.member.updated", "group.role.updated", "group.mute.updated":
if _, ok := copyPayload["user_id"]; ok {
copyPayload["member_sync"] = "single"
} else {
copyPayload["member_sync"] = "full"
}
}
if isAdminOnlyIMGroupEvent(eventType) {
copyPayload["visibility"] = "admins_and_affected"
} else if eventType == "group.member.joined" {
copyPayload["visibility"] = "all_members"
}
return copyPayload
}
func userIDs(uids []types.Uid) []string { func userIDs(uids []types.Uid) []string {
ids := make([]string, 0, len(uids)) ids := make([]string, 0, len(uids))
for _, uid := range uids { for _, uid := range uids {
+149
View File
@@ -0,0 +1,149 @@
package main
import (
"encoding/json"
"testing"
"time"
"github.com/golang/mock/gomock"
"github.com/tinode/chat/server/store"
"github.com/tinode/chat/server/store/mock_store"
"github.com/tinode/chat/server/store/types"
)
func TestPublishIMGroupRealtimeEventRestrictsMemberRemoval(t *testing.T) {
oldGroupBusiness := store.GroupBusiness
oldSessions := globals.sessionStore
oldHub := globals.hub
admin := types.Uid(1)
affected := types.Uid(2)
regular := types.Uid(3)
actor := types.Uid(4)
store.GroupBusiness = &testGroupBusinessStore{listMembers: []types.GroupMember{
{UserID: admin, Role: types.GroupMemberRoleAdmin, MemberStatus: types.GroupMemberStatusActive},
}}
globals.sessionStore = NewSessionStore(time.Minute)
globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 2)}
t.Cleanup(func() {
store.GroupBusiness = oldGroupBusiness
globals.sessionStore = oldSessions
globals.hub = oldHub
})
sessions := map[types.Uid]*Session{}
for _, uid := range []types.Uid{admin, affected, regular} {
sess := &Session{
sid: uid.UserId(), uid: uid, tenantID: 7, proto: WEBSOCK,
send: make(chan any, 2), subs: make(map[string]*Subscription),
}
globals.sessionStore.sessCache[sess.sid] = sess
sessions[uid] = sess
}
publishIMGroupRealtimeEvent(7, "grpRestricted", actor, "group.member.left",
map[string]any{"user_id": affected.UserId()})
for _, uid := range []types.Uid{admin, affected} {
select {
case raw := <-sessions[uid].send:
msg, ok := raw.(*ServerComMessage)
if !ok || msg.Info == nil || msg.Info.Event != "group.member.left" || msg.Info.Topic != "grpRestricted" {
t.Fatalf("unexpected event for %s: %#v", uid.UserId(), raw)
}
var payload map[string]any
if err := json.Unmarshal(msg.Info.Payload, &payload); err != nil {
t.Fatal(err)
}
if payload["member_sync"] != "remove" || payload["visibility"] != "admins_and_affected" {
t.Fatalf("unexpected payload for %s: %#v", uid.UserId(), payload)
}
default:
t.Fatalf("expected event for %s", uid.UserId())
}
}
select {
case raw := <-sessions[regular].send:
t.Fatalf("regular member received restricted event: %#v", raw)
default:
}
if len(globals.hub.routeSrv) != 0 {
t.Fatalf("restricted event was broadcast to topic")
}
}
func TestPublishIMGroupRealtimeEventBroadcastsMemberJoin(t *testing.T) {
oldHub := globals.hub
globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 1)}
t.Cleanup(func() { globals.hub = oldHub })
publishIMGroupRealtimeEvent(7, "grpPublic", types.Uid(1), "group.member.joined",
map[string]any{"user_id": types.Uid(2).UserId()})
select {
case msg := <-globals.hub.routeSrv:
if msg.Info == nil || msg.Info.Event != "group.member.joined" {
t.Fatalf("unexpected joined event: %#v", msg)
}
var payload map[string]any
if err := json.Unmarshal(msg.Info.Payload, &payload); err != nil {
t.Fatal(err)
}
if payload["member_sync"] != "single" || payload["visibility"] != "all_members" {
t.Fatalf("unexpected joined payload: %#v", payload)
}
default:
t.Fatal("joined event was not broadcast")
}
}
func TestSendPushForStoredGroupMessageIncludesEverySubscription(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
oldTopics := store.Topics
oldUsersUpdate := globals.usersUpdate
oldCluster := globals.cluster
topics := mock_store.NewMockTopicsPersistenceInterface(ctrl)
store.Topics = topics
globals.usersUpdate = make(chan *UserCacheReq, 1)
globals.cluster = nil
t.Cleanup(func() {
store.Topics = oldTopics
globals.usersUpdate = oldUsersUpdate
globals.cluster = oldCluster
})
subs := make([]types.Subscription, 0, 1500)
for i := 1; i <= 1500; i++ {
sub := types.Subscription{
User: types.Uid(i).String(), Topic: "grpLarge", ModeWant: types.ModeCFull, ModeGiven: types.ModeCFull,
}
if i == 1500 {
sub.Private = map[string]any{"mute_notifications": true}
}
subs = append(subs, sub)
}
topics.EXPECT().GetSubs(types.TenantID(7), "grpLarge", nil).Return(subs, nil)
msg := &types.Message{
ObjHeader: types.ObjHeader{CreatedAt: time.Now().UTC()}, TenantID: 7,
Topic: "grpLarge", SeqId: 99, Content: map[string]any{"text": "成员加入了群聊"},
}
sendPushForStoredGroupMessage(7, msg)
select {
case req := <-globals.usersUpdate:
if req.PushRcpt == nil || len(req.PushRcpt.To) != len(subs) {
t.Fatalf("push recipients: got %d want %d", len(req.PushRcpt.To), len(subs))
}
for _, recipient := range req.PushRcpt.To {
if !recipient.ShouldIncrementUnreadCountInCache {
t.Fatal("system group message did not increment per-user unread")
}
}
if !req.PushRcpt.To[types.Uid(1500)].Silent {
t.Fatal("muted group recipient must receive a silent push")
}
default:
t.Fatal("group system message did not enqueue push")
}
}
+63
View File
@@ -3967,6 +3967,15 @@ func serveIMGroupResource(wrt http.ResponseWriter, req *http.Request) {
serveIMGroupMembers(wrt, req, topic) serveIMGroupMembers(wrt, req, topic)
return return
} }
if strings.HasPrefix(tail, "members/") && !strings.HasSuffix(tail, "/profile") {
memberID := types.ParseUserId(strings.TrimSpace(strings.TrimPrefix(tail, "members/")))
if memberID.IsZero() {
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "group route not found", nil)
return
}
serveIMGroupMember(wrt, req, topic, memberID)
return
}
if strings.HasPrefix(tail, "members/") && strings.HasSuffix(tail, "/profile") { if strings.HasPrefix(tail, "members/") && strings.HasSuffix(tail, "/profile") {
memberID, ok := groupMemberProfilePath(tail) memberID, ok := groupMemberProfilePath(tail)
if !ok { if !ok {
@@ -4066,11 +4075,60 @@ func serveIMGroupMembers(wrt http.ResponseWriter, req *http.Request, topic strin
"topic": topic, "topic": topic,
"items": items, "items": items,
"next_cursor": nextCursor, "next_cursor": nextCursor,
"sync_mode": "full",
"permissions": permissions, "permissions": permissions,
} }
writeIMSuccess(wrt, req, payload) writeIMSuccess(wrt, req, payload)
} }
func serveIMGroupMember(wrt http.ResponseWriter, req *http.Request, topic string, memberID types.Uid) {
if req.Method != http.MethodGet {
writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
return
}
principal, selfSub, ok := requireIMGroupMember(wrt, req, topic, true)
if !ok {
return
}
if !(selfSub.ModeGiven & selfSub.ModeWant).IsReader() {
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "topic read permission required", nil)
return
}
member, err := store.GroupBusiness.GetMember(principal.TenantID, topic, memberID)
if err != nil {
if errors.Is(err, types.ErrNotFound) {
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "member not found", nil)
return
}
writeIMRelationError(wrt, req, err)
return
}
if member == nil || member.MemberStatus != types.GroupMemberStatusActive {
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "member not found", nil)
return
}
memberSub, err := store.Subs.Get(principal.TenantID, topic, memberID, false)
if err != nil {
writeIMRelationError(wrt, req, err)
return
}
if memberSub == nil || !(memberSub.ModeGiven & memberSub.ModeWant).IsJoiner() {
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "member not found", nil)
return
}
member.ModeWant = memberSub.ModeWant
member.ModeGiven = memberSub.ModeGiven
if user, userErr := store.Users.Get(principal.TenantID, memberID); userErr == nil && user != nil {
member.Public = user.Public
member.Trusted = user.Trusted
}
remarks := groupMemberRemarkMap(principal.TenantID, principal.Uid, []types.GroupMember{*member})
payload := groupBusinessMemberPayload(*member, remarks[member.UserID])
payload["topic"] = topic
payload["sync_mode"] = "single"
writeIMSuccess(wrt, req, payload)
}
func serveIMGroupDetail(wrt http.ResponseWriter, req *http.Request, topic string) { func serveIMGroupDetail(wrt http.ResponseWriter, req *http.Request, topic string) {
principal, selfSub, ok := requireIMGroupMember(wrt, req, topic, true) principal, selfSub, ok := requireIMGroupMember(wrt, req, topic, true)
if !ok { if !ok {
@@ -6861,6 +6919,10 @@ func groupBusinessMemberPayload(member types.GroupMember, remark string) map[str
data["group_nickname"] = nickname data["group_nickname"] = nickname
} }
data["role"] = groupMemberRole(access) data["role"] = groupMemberRole(access)
data["member_status"] = member.MemberStatus
data["version"] = member.Version
data["joined_at"] = member.JoinedAt.UTC().Format(time.RFC3339Nano)
data["updated_at"] = member.UpdatedAt.UTC().Format(time.RFC3339Nano)
data["mentionable"] = access.IsJoiner() && access.IsReader() data["mentionable"] = access.IsJoiner() && access.IsReader()
data["acs"] = map[string]any{ data["acs"] = map[string]any{
"mode": access.String(), "mode": access.String(),
@@ -9941,6 +10003,7 @@ func publishIMConversationSystemNotice(tenantID types.TenantID, topic, msgType,
return return
} }
broadcastIMConversationSystemMessage(tenantID, msg, now) broadcastIMConversationSystemMessage(tenantID, msg, now)
sendPushForStoredGroupMessage(tenantID, msg)
} }
func broadcastIMConversationSystemMessage(tenantID types.TenantID, msg *types.Message, timestamp time.Time) { func broadcastIMConversationSystemMessage(tenantID types.TenantID, msg *types.Message, timestamp time.Time) {
+56
View File
@@ -339,6 +339,9 @@ func (s *testGroupBusinessStore) UpsertMember(_ types.TenantID, member *types.Gr
} }
func (s *testGroupBusinessStore) GetMember(_ types.TenantID, topic string, uid types.Uid) (*types.GroupMember, error) { func (s *testGroupBusinessStore) GetMember(_ types.TenantID, topic string, uid types.Uid) (*types.GroupMember, error) {
if s.member != nil {
return s.member, nil
}
return &types.GroupMember{Topic: topic, UserID: uid, Role: types.GroupMemberRoleMember, MemberStatus: types.GroupMemberStatusActive, Version: 1}, nil return &types.GroupMember{Topic: topic, UserID: uid, Role: types.GroupMemberRoleMember, MemberStatus: types.GroupMemberStatusActive, Version: 1}, nil
} }
@@ -3034,6 +3037,59 @@ func TestServeIMGroupMembers(t *testing.T) {
} }
} }
func TestServeIMGroupMemberReturnsSingleSnapshot(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
expectIMBearer(t, ctrl, "token-1", 7, 42)
oldSubs := store.Subs
oldUsers := store.Users
oldGroupBusiness := store.GroupBusiness
oldRelations := store.Relations
subs := mock_store.NewMockSubsPersistenceInterface(ctrl)
users := mock_store.NewMockUsersPersistenceInterface(ctrl)
store.Subs = subs
store.Users = users
memberID := types.Uid(101)
joinedAt := time.Date(2026, 9, 18, 0, 0, 0, 0, time.UTC)
store.GroupBusiness = &testGroupBusinessStore{member: &types.GroupMember{
ID: 101, Topic: "grpMention", UserID: memberID, Role: types.GroupMemberRoleMember,
MemberStatus: types.GroupMemberStatusActive, Nickname: "群里小王", Version: 3,
JoinedAt: joinedAt, UpdatedAt: joinedAt.Add(time.Minute),
}}
store.Relations = &testRelationsStore{context: &types.RelationshipContext{
TenantID: 7, UserID: 42, PeerID: memberID, Relationship: types.RelationshipNone,
}}
t.Cleanup(func() {
store.Subs = oldSubs
store.Users = oldUsers
store.GroupBusiness = oldGroupBusiness
store.Relations = oldRelations
})
fullSub := func(uid types.Uid) *types.Subscription {
return &types.Subscription{TenantID: 7, User: uid.String(), Topic: "grpMention", ModeWant: types.ModeCFull, ModeGiven: types.ModeCFull}
}
subs.EXPECT().Get(types.TenantID(7), "grpMention", types.Uid(42), false).Return(fullSub(42), nil)
subs.EXPECT().Get(types.TenantID(7), "grpMention", memberID, false).Return(fullSub(memberID), nil)
users.EXPECT().Get(types.TenantID(7), memberID).Return(&types.User{
ObjHeader: types.ObjHeader{Id: memberID.String()}, Public: map[string]any{"nickname": "Alice", "avatar_file_id": "alice.jpg"},
}, nil)
req := httptest.NewRequest(http.MethodGet, "/v1/im/groups/grpMention/members/"+memberID.UserId(), nil)
req.Header.Set("Authorization", "Bearer token-1")
rec := httptest.NewRecorder()
serveIMGroupResource(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
data := decodeIMHTTPResponse(t, rec).Data.(map[string]any)
if data["user_id"] != memberID.UserId() || data["sync_mode"] != "single" || data["version"].(float64) != 3 ||
data["group_nickname"] != "群里小王" || data["nickname"] != "Alice" {
t.Fatalf("unexpected single member response: %#v", data)
}
}
func TestGroupMemberPayloadUserIDFormats(t *testing.T) { func TestGroupMemberPayloadUserIDFormats(t *testing.T) {
raw := types.Uid(101) raw := types.Uid(101)
fromRaw := groupMemberPayload(types.Subscription{ fromRaw := groupMemberPayload(types.Subscription{
+56
View File
@@ -91,6 +91,8 @@ func (t *Topic) pushForData(fromUid types.Uid, data *MsgServerData, msgMarkedAsR
// Number of attached sessions the data message will be delivered to. // Number of attached sessions the data message will be delivered to.
// Push notifications sent to users with non-zero online sessions will be marked silent. // Push notifications sent to users with non-zero online sessions will be marked silent.
Delivered: online, Delivered: online,
Silent: t.cat == types.TopicCatGrp &&
boolFromAny(groupPrivateMap(pud.private)["mute_notifications"]),
// Unread counts are incremented for all recipients, // Unread counts are incremented for all recipients,
// and for sender only if the message wasnt't marked 'read' by the sender // and for sender only if the message wasnt't marked 'read' by the sender
ShouldIncrementUnreadCountInCache: uid != fromUid || !msgMarkedAsReadBySender, ShouldIncrementUnreadCountInCache: uid != fromUid || !msgMarkedAsReadBySender,
@@ -104,6 +106,60 @@ func (t *Topic) pushForData(fromUid types.Uid, data *MsgServerData, msgMarkedAsR
return nil return nil
} }
// sendPushForStoredGroupMessage handles server-generated group messages which
// bypass Topic.handlePubBroadcast but must still update per-user unread counts
// and notify offline devices.
func sendPushForStoredGroupMessage(tenantID types.TenantID, msg *types.Message) {
if msg == nil || globals.usersUpdate == nil || types.GetTopicCat(msg.Topic) != types.TopicCatGrp || store.Topics == nil {
return
}
subs, err := store.Topics.GetSubs(tenantID, msg.Topic, nil)
if err != nil {
logs.Warn.Printf("group system push: load recipients topic=%s: %v", msg.Topic, err)
return
}
receipt := &push.Receipt{
TenantID: tenantID,
To: make(map[types.Uid]push.Recipient, len(subs)),
Payload: push.Payload{
What: push.ActMsg, Topic: msg.Topic, Timestamp: msg.CreatedAt, SeqId: msg.SeqId,
Content: msg.Content,
},
}
if mime, ok := msg.Head["mime"].(string); ok {
receipt.Payload.ContentType = mime
}
for _, sub := range subs {
uid := types.ParseUid(sub.User)
mode := sub.ModeWant & sub.ModeGiven
if uid.IsZero() || !mode.IsPresencer() || !mode.IsReader() {
continue
}
delivered := 0
if globals.sessionStore != nil {
globals.sessionStore.Range(func(_ string, sess *Session) bool {
if sess != nil && !sess.isCluster() && !sess.isProxy() && sess.tenantID == tenantID && sess.uid == uid {
delivered++
}
return true
})
}
if presence, presenceErr := store.GetUserPresence(tenantID, uid); presenceErr != nil {
logs.Warn.Printf("group system push: read presence topic=%s uid=%s: %v", msg.Topic, uid.UserId(), presenceErr)
} else if presence != nil && presence.Online && presence.SessionCount > delivered {
delivered = presence.SessionCount
}
receipt.To[uid] = push.Recipient{
Delivered: delivered,
Silent: boolFromAny(groupPrivateMap(sub.Private)["mute_notifications"]),
ShouldIncrementUnreadCountInCache: true,
}
}
if len(receipt.To) > 0 {
sendPush(receipt)
}
}
func (t *Topic) preparePushForSubReceipt(fromUid types.Uid, now time.Time) *push.Receipt { func (t *Topic) preparePushForSubReceipt(fromUid types.Uid, now time.Time) *push.Receipt {
// P2P push payloads use the canonical p2p... topic. usr... is accepted only // P2P push payloads use the canonical p2p... topic. usr... is accepted only
// as an input alias. // as an input alias.
+1 -1
View File
@@ -158,7 +158,7 @@ func PrepareV1Notifications(rcpt *push.Receipt, config *configType) ([]*fcmv1.Me
for uid, devList := range devices { for uid, devList := range devices {
topic := rcpt.Payload.Topic topic := rcpt.Payload.Topic
userData := data userData := data
if rcpt.To[uid].Delivered > 0 { if rcpt.To[uid].Delivered > 0 || rcpt.To[uid].Silent {
userData = clonePayload(data) userData = clonePayload(data)
// Silence the push for user who have received the data interactively. // Silence the push for user who have received the data interactively.
userData["silent"] = "true" userData["silent"] = "true"
+1 -1
View File
@@ -351,7 +351,7 @@ func preparePayloads(rcpt *push.Receipt, config *configType) ([]*jpushPayload, [
for _, uid := range uids { for _, uid := range uids {
recipient := rcpt.To[uid] recipient := rcpt.To[uid]
userData := cloneData(data) userData := cloneData(data)
if recipient.Delivered > 0 { if recipient.Delivered > 0 || recipient.Silent {
userData["silent"] = "true" userData["silent"] = "true"
} }
topic := rcpt.Payload.Topic topic := rcpt.Payload.Topic
+2
View File
@@ -28,6 +28,8 @@ type Recipient struct {
Delivered int `json:"delivered"` Delivered int `json:"delivered"`
// List of user's devices that the packet was delivered to (if known). Len(Devices) >= Delivered // List of user's devices that the packet was delivered to (if known). Len(Devices) >= Delivered
Devices []string `json:"devices,omitempty"` Devices []string `json:"devices,omitempty"`
// Deliver as a data-only notification without showing an alert.
Silent bool `json:"silent,omitempty"`
// Unread count to include in the push // Unread count to include in the push
Unread int `json:"unread"` Unread int `json:"unread"`
// Indicates whether unread counter in the cache should be incremented before sending the push. // Indicates whether unread counter in the cache should be incremented before sending the push.
+33
View File
@@ -564,6 +564,39 @@ func TestDispatchAlreadySubscribed(t *testing.T) {
verifyResponseCodes(&r, []int{http.StatusNotModified}, t) verifyResponseCodes(&r, []int{http.StatusNotModified}, t)
} }
func TestSubscribeP2PUserAliasReachesTopic(t *testing.T) {
oldHub := globals.hub
globals.hub = &Hub{join: make(chan *ClientComMessage, 1)}
t.Cleanup(func() { globals.hub = oldHub })
self := types.Uid(1)
peer := types.Uid(2)
canonical := self.P2PName(peer)
s := test_makeSession(self)
s.subs = map[string]*Subscription{canonical: {}}
msg := &ClientComMessage{
Sub: &MsgClientSub{Id: "alias-sub", Topic: peer.UserId()},
Id: "alias-sub",
Original: peer.UserId(),
AsUser: self.UserId(),
TenantID: s.tenantID,
Timestamp: types.TimeNow(),
sess: s,
}
s.subscribe(msg)
select {
case routed := <-globals.hub.join:
if routed != msg || routed.RcptTo != canonical {
t.Fatalf("unexpected alias subscription route: %#v", routed)
}
s.inflightReqs.Done()
default:
t.Fatal("p2p user alias was incorrectly handled as an already-subscribed topic")
}
}
func TestDispatchSubscribeJoinChannelFull(t *testing.T) { func TestDispatchSubscribeJoinChannelFull(t *testing.T) {
uid := types.Uid(1) uid := types.Uid(1)
s := test_makeSession(uid) s := test_makeSession(uid)