From c6e7f6b923e6416635d8f6c5b9dd03b7a3859f13 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Fri, 18 Sep 2026 00:58:05 +0800 Subject: [PATCH] feat: improve group member events and push delivery --- docs/IM-前后端对接文档.md | 11 ++- docs/IM-服务端主动推送消息梳理.md | 18 ++-- server/db/mysql/adapter.go | 9 +- server/db/mysql/tenant_adapter.go | 9 +- server/im_group_events.go | 130 ++++++++++++++++++++++++++ server/im_group_events_test.go | 149 ++++++++++++++++++++++++++++++ server/im_http.go | 63 +++++++++++++ server/im_http_test.go | 56 +++++++++++ server/push.go | 56 +++++++++++ server/push/fcm/payload.go | 2 +- server/push/jpush/push_jpush.go | 2 +- server/push/push.go | 2 + server/session_test.go | 33 +++++++ 13 files changed, 521 insertions(+), 19 deletions(-) create mode 100644 server/im_group_events_test.go diff --git a/docs/IM-前后端对接文档.md b/docs/IM-前后端对接文档.md index 8674c67..873c932 100644 --- a/docs/IM-前后端对接文档.md +++ b/docs/IM-前后端对接文档.md @@ -438,7 +438,9 @@ X-MeetingIM-APIKey: <预置的普通 API Key> "event": "group.member.joined", "payload": { "topic": "grp_xxx", - "user_id": "usr_xxx" + "user_id": "usr_xxx", + "member_sync": "single", + "visibility": "all_members" }, "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.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` 做补偿同步 | -| 群业务变化 | 群 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_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"}` | | `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"}` | -| `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}}` | | `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"}]}` | diff --git a/docs/IM-服务端主动推送消息梳理.md b/docs/IM-服务端主动推送消息梳理.md index f0828e1..8605e74 100644 --- a/docs/IM-服务端主动推送消息梳理.md +++ b/docs/IM-服务端主动推送消息梳理.md @@ -121,19 +121,21 @@ | --- | --- | --- | --- | --- | | 创建群 | 群成员 | `info.what="group"`, `event="group.created"` | `topic`、群基础信息 | 除操作人外写入 | | 更新群设置 | 群成员 | `event="group.settings.updated"` | `topic` | 除操作人外写入 | -| 成员资料/权限更新 | 群成员 | `event="group.member.updated"` | `user_id` | 除操作人外写入 | -| 成员加入 | 群成员 | `event="group.member.joined"` | `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.member.updated"` | `user_id`、`member_sync=single` | 不写群系统消息 | +| 成员加入 | 群成员 | `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.accepted/rejected"` | `request_id`、`status`、`topic` | 普通成员不可见；接受时另发全群加入消息 | | 群公告更新 | 群成员 | `event="group.announcement.updated"` | `topic`、公告信息 | 除操作人外写入 | -| 群禁言更新 | 群成员 | `event="group.mute.updated"` | `topic`、禁言信息 | 除操作人外写入 | -| 成员离群 | 群成员 | `event="group.member.left"` | `user_id` | 除操作人外写入 | -| 群角色更新 | 群成员 | `event="group.role.updated"` | `user_id`、角色信息 | 除操作人外写入 | +| 群禁言更新 | 群主、管理员、被操作人 | `event="group.mute.updated"` | `topic`、禁言信息、`member_sync` | 普通成员不可见 | +| 成员离群/移除 | 群主、管理员、离群成员 | `event="group.member.left"` | `user_id`、`member_sync=remove` | 普通成员不可见 | +| 群角色更新 | 群主、管理员、被操作人 | `event="group.role.updated"` | `user_id`、角色信息、`member_sync` | 普通成员不可见 | | 群主转让 | 群成员 | `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 待处理数。 - 群公告变化：刷新公告区。 - 成员加入/离开：刷新群成员数、成员列表和会话系统提示。 diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 8e2f97b..abe01e2 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -2950,7 +2950,7 @@ func (a *adapter) SubsForTopic(topic string, keepDeleted bool, opts *t.QueryOpt) // Filter out deleted rows. q += " AND deletedat IS NULL" } - limit := a.maxResults + limit := 0 if opts != nil { // Ignore IfModifiedSince - we must return all entries // 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=?" args = append(args, store.DecodeUid(opts.User)) } + limit = a.maxResults if opts.Limit > 0 && opts.Limit < limit { limit = opts.Limit } } - q += " LIMIT ?" - args = append(args, limit) + if limit > 0 { + q += " LIMIT ?" + args = append(args, limit) + } ctx, cancel := a.getContext() if cancel != nil { diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 9cb5610..dc6622c 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -1779,8 +1779,9 @@ func (a *adapter) TenantSubsForTopic(id t.TenantID, topic string, keepDeleted bo if !keepDeleted { q += " AND deletedat IS NULL" } - limit := a.maxResults + limit := 0 if opts != nil { + limit = a.maxResults if !opts.User.IsZero() { q += " AND userid=?" args = append(args, decodeUID(opts.User)) @@ -1789,8 +1790,10 @@ func (a *adapter) TenantSubsForTopic(id t.TenantID, topic string, keepDeleted bo limit = opts.Limit } } - q += " LIMIT ?" - args = append(args, limit) + if limit > 0 { + q += " LIMIT ?" + args = append(args, limit) + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() diff --git a/server/im_group_events.go b/server/im_group_events.go index a8397e6..d1cca39 100644 --- a/server/im_group_events.go +++ b/server/im_group_events.go @@ -2,10 +2,12 @@ package main import ( "encoding/json" + "fmt" "strings" "time" "github.com/tinode/chat/server/logs" + "github.com/tinode/chat/server/store" "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 == "" { return } + payload = enrichIMGroupMemberEventPayload(eventType, payload) encoded, err := json.Marshal(payload) if err != nil { 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() { from = actor.UserId() } + if isAdminOnlyIMGroupEvent(eventType) { + publishIMGroupEventToAdmins(tenantID, topic, actor, eventType, encoded, payload) + return + } message := &ServerComMessage{ TenantID: tenantID, RcptTo: topic, @@ -47,6 +54,129 @@ func publishIMGroupRealtimeEvent(tenantID types.TenantID, topic string, actor ty 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 { ids := make([]string, 0, len(uids)) for _, uid := range uids { diff --git a/server/im_group_events_test.go b/server/im_group_events_test.go new file mode 100644 index 0000000..6a1368a --- /dev/null +++ b/server/im_group_events_test.go @@ -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") + } +} diff --git a/server/im_http.go b/server/im_http.go index 6ea3d29..572d516 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -3967,6 +3967,15 @@ func serveIMGroupResource(wrt http.ResponseWriter, req *http.Request) { serveIMGroupMembers(wrt, req, topic) 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") { memberID, ok := groupMemberProfilePath(tail) if !ok { @@ -4066,11 +4075,60 @@ func serveIMGroupMembers(wrt http.ResponseWriter, req *http.Request, topic strin "topic": topic, "items": items, "next_cursor": nextCursor, + "sync_mode": "full", "permissions": permissions, } 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) { principal, selfSub, ok := requireIMGroupMember(wrt, req, topic, true) if !ok { @@ -6861,6 +6919,10 @@ func groupBusinessMemberPayload(member types.GroupMember, remark string) map[str data["group_nickname"] = nickname } 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["acs"] = map[string]any{ "mode": access.String(), @@ -9941,6 +10003,7 @@ func publishIMConversationSystemNotice(tenantID types.TenantID, topic, msgType, return } broadcastIMConversationSystemMessage(tenantID, msg, now) + sendPushForStoredGroupMessage(tenantID, msg) } func broadcastIMConversationSystemMessage(tenantID types.TenantID, msg *types.Message, timestamp time.Time) { diff --git a/server/im_http_test.go b/server/im_http_test.go index 6606ec5..82cb4d6 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -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) { + if s.member != nil { + return s.member, 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) { raw := types.Uid(101) fromRaw := groupMemberPayload(types.Subscription{ diff --git a/server/push.go b/server/push.go index 9581a27..50529eb 100644 --- a/server/push.go +++ b/server/push.go @@ -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. // Push notifications sent to users with non-zero online sessions will be marked silent. Delivered: online, + Silent: t.cat == types.TopicCatGrp && + boolFromAny(groupPrivateMap(pud.private)["mute_notifications"]), // Unread counts are incremented for all recipients, // and for sender only if the message wasnt't marked 'read' by the sender ShouldIncrementUnreadCountInCache: uid != fromUid || !msgMarkedAsReadBySender, @@ -104,6 +106,60 @@ func (t *Topic) pushForData(fromUid types.Uid, data *MsgServerData, msgMarkedAsR 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 { // P2P push payloads use the canonical p2p... topic. usr... is accepted only // as an input alias. diff --git a/server/push/fcm/payload.go b/server/push/fcm/payload.go index f45169a..5425698 100644 --- a/server/push/fcm/payload.go +++ b/server/push/fcm/payload.go @@ -158,7 +158,7 @@ func PrepareV1Notifications(rcpt *push.Receipt, config *configType) ([]*fcmv1.Me for uid, devList := range devices { topic := rcpt.Payload.Topic userData := data - if rcpt.To[uid].Delivered > 0 { + if rcpt.To[uid].Delivered > 0 || rcpt.To[uid].Silent { userData = clonePayload(data) // Silence the push for user who have received the data interactively. userData["silent"] = "true" diff --git a/server/push/jpush/push_jpush.go b/server/push/jpush/push_jpush.go index 34af440..007955a 100644 --- a/server/push/jpush/push_jpush.go +++ b/server/push/jpush/push_jpush.go @@ -351,7 +351,7 @@ func preparePayloads(rcpt *push.Receipt, config *configType) ([]*jpushPayload, [ for _, uid := range uids { recipient := rcpt.To[uid] userData := cloneData(data) - if recipient.Delivered > 0 { + if recipient.Delivered > 0 || recipient.Silent { userData["silent"] = "true" } topic := rcpt.Payload.Topic diff --git a/server/push/push.go b/server/push/push.go index 410e681..400d40a 100644 --- a/server/push/push.go +++ b/server/push/push.go @@ -28,6 +28,8 @@ type Recipient struct { Delivered int `json:"delivered"` // List of user's devices that the packet was delivered to (if known). Len(Devices) >= Delivered 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 int `json:"unread"` // Indicates whether unread counter in the cache should be incremented before sending the push. diff --git a/server/session_test.go b/server/session_test.go index 3105016..c0f4cd4 100644 --- a/server/session_test.go +++ b/server/session_test.go @@ -564,6 +564,39 @@ func TestDispatchAlreadySubscribed(t *testing.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) { uid := types.Uid(1) s := test_makeSession(uid)