From 7b304412b0347bc08015cd19b0003a4533243f35 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sun, 13 Sep 2026 16:03:46 +0800 Subject: [PATCH] fix: repair group management edge cases --- server/db/mysql/tenant_groups.go | 215 ++++++++++++-------------- server/im_http.go | 185 +++++++++++++++++----- server/im_http_test.go | 257 +++++++++++++++++++++++++++---- 3 files changed, 473 insertions(+), 184 deletions(-) diff --git a/server/db/mysql/tenant_groups.go b/server/db/mysql/tenant_groups.go index 140bf83..fb3b253 100644 --- a/server/db/mysql/tenant_groups.go +++ b/server/db/mysql/tenant_groups.go @@ -133,6 +133,14 @@ func (a *adapter) TenantGroupSettingsUpdate(id t.TenantID, topic string, patch t set = append(set, "member_invite=?") args = append(args, *patch.MemberInvite) } + if patch.MaxMemberCount != nil { + set = append(set, "max_member_count=?") + args = append(args, *patch.MaxMemberCount) + } + if patch.MaxAdminCount != nil { + set = append(set, "max_admin_count=?") + args = append(args, *patch.MaxAdminCount) + } if patch.MemberCountDelta != 0 { set = append(set, "member_count=GREATEST(member_count+?,0)") args = append(args, patch.MemberCountDelta) @@ -163,9 +171,42 @@ func (a *adapter) TenantGroupSettingsUpdate(id t.TenantID, topic string, patch t } return nil, t.ErrNotFound } - if err = insertGroupEventTx(tx, topic, t.ZeroUid, "group.settings.updated", map[string]any{"topic": topic}); err != nil { + if err = tx.Commit(); err != nil { return nil, err } + return a.groupSettingsByTopic(ctx, topic) +} + +func (a *adapter) TenantGroupDissolve(id t.TenantID, topic string, actor t.Uid, reason string, now time.Time) (*t.GroupSettings, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupDissolve(id, topic, actor, reason, now) + } + if strings.TrimSpace(topic) == "" || actor.IsZero() { + return nil, t.ErrMalformed + } + if now.IsZero() { + now = time.Now().UTC().Round(time.Millisecond) + } + ctx, cancel := a.getContextForTx() + if cancel != nil { + defer cancel() + } + tx, err := a.db.BeginTxx(ctx, nil) + if err != nil { + return nil, err + } + defer tx.Rollback() + res, err := tx.ExecContext(ctx, `UPDATE im_group_settings + SET group_status='dissolved',dissolved_by=?,dissolved_at=?,version=version+1,updated_at=? + WHERE topic=? AND group_status<>'dissolved'`, decodeUID(actor), now, now, strings.TrimSpace(topic)) + if err != nil { + return nil, err + } + if affected, _ := res.RowsAffected(); affected == 0 { + return nil, t.ErrNotFound + } if err = tx.Commit(); err != nil { return nil, err } @@ -391,9 +432,6 @@ func (a *adapter) TenantGroupMemberUpdate(id t.TenantID, topic string, uid t.Uid if err = recountGroupMembersTx(tx, topic); err != nil { return nil, err } - if err = insertGroupEventTx(tx, topic, uid, "group.member.updated", map[string]any{"user_id": uid.UserId()}); err != nil { - return nil, err - } if err = tx.Commit(); err != nil { return nil, err } @@ -423,6 +461,21 @@ func (a *adapter) TenantGroupInvitationCreate(id t.TenantID, inv *t.GroupInvitat if cancel != nil { defer cancel() } + if strings.TrimSpace(inv.JoinRequestID) != "" { + existing, err := a.groupInvitationByQuery(ctx, `SELECT inv.id,inv.invitation_id,inv.topic,inv.inviter_user_id,inv.target_user_id,inv.message,inv.status,inv.result, + inv.join_request_id,inv.idempotency_key,inv.expires_at,inv.handled_at,inv.created_at,inv.updated_at + FROM im_group_invitations AS inv + INNER JOIN im_group_join_requests AS req ON req.request_id=inv.join_request_id + WHERE inv.topic=? AND inv.target_user_id=? AND inv.status=? AND req.status=? + ORDER BY req.id DESC LIMIT 1`, + inv.Topic, decodeUID(inv.TargetUserID), t.GroupInvitationStatusPending, t.GroupJoinRequestStatusPending) + if err == nil { + return existing, nil + } + if err != t.ErrNotFound { + return nil, err + } + } tx, err := a.db.BeginTxx(ctx, nil) if err != nil { return nil, err @@ -441,11 +494,6 @@ func (a *adapter) TenantGroupInvitationCreate(id t.TenantID, inv *t.GroupInvitat if err != nil { return nil, err } - if err = insertGroupEventTx(tx, inv.Topic, inv.TargetUserID, "group.join_request.created", map[string]any{ - "request_id": inv.JoinRequestID, "applicant_user_id": inv.TargetUserID.UserId(), "source": "invitation", "invitation_id": inv.InvitationID, - }); err != nil { - return nil, err - } } if err = tx.Commit(); err != nil { return nil, err @@ -544,11 +592,6 @@ func (a *adapter) TenantGroupInvitationAccept(id t.TenantID, invitationID string return nil, err } } - if err = insertGroupEventTx(tx, inv.Topic, uid, "group.member.joined", map[string]any{ - "user_id": uid.UserId(), "source": "invitation", "invitation_id": inv.InvitationID, - }); err != nil { - return nil, err - } if err = tx.Commit(); err != nil { return nil, err } @@ -582,6 +625,14 @@ func (a *adapter) TenantGroupJoinRequestCreate(id t.TenantID, req *t.GroupJoinRe if cancel != nil { defer cancel() } + existing, err := a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE topic=? AND applicant_user_id=? AND status=? ORDER BY id DESC LIMIT 1", + req.Topic, decodeUID(req.ApplicantUserID), t.GroupJoinRequestStatusPending) + if err == nil { + return existing, nil + } + if err != t.ErrNotFound { + return nil, err + } tx, err := a.db.BeginTxx(ctx, nil) if err != nil { return nil, err @@ -593,11 +644,6 @@ func (a *adapter) TenantGroupJoinRequestCreate(id t.TenantID, req *t.GroupJoinRe if err != nil { return nil, err } - if err = insertGroupEventTx(tx, req.Topic, req.ApplicantUserID, "group.join_request.created", map[string]any{ - "request_id": req.RequestID, "applicant_user_id": req.ApplicantUserID.UserId(), - }); err != nil { - return nil, err - } if err = tx.Commit(); err != nil { return nil, err } @@ -610,18 +656,26 @@ func (a *adapter) TenantGroupJoinRequestList(id t.TenantID, topic, status, curso } else if tenantAdp != a { return tenantAdp.TenantGroupJoinRequestList(id, topic, status, cursor, limit) } - where := []string{"topic=?"} + where := []string{"r.topic=?"} args := []any{topic} if status != "" && status != "all" { - where = append(where, "status=?") + where = append(where, "r.status=?") args = append(args, status) + if status == t.GroupJoinRequestStatusPending { + where = append(where, `NOT EXISTS ( + SELECT 1 FROM im_group_join_requests AS newer + WHERE newer.topic=r.topic AND newer.applicant_user_id=r.applicant_user_id + AND newer.status=? AND newer.id>r.id + )`) + args = append(args, status) + } } if cursor != "" { parsed, err := strconv.ParseInt(cursor, 10, 64) if err != nil || parsed <= 0 { return nil, "", t.ErrMalformed } - where = append(where, "id>?") + where = append(where, "r.id>?") args = append(args, parsed) } args = append(args, limit+1) @@ -629,7 +683,7 @@ func (a *adapter) TenantGroupJoinRequestList(id t.TenantID, topic, status, curso if cancel != nil { defer cancel() } - rows, err := a.db.QueryxContext(ctx, "SELECT * FROM im_group_join_requests WHERE "+strings.Join(where, " AND ")+" ORDER BY id LIMIT ?", args...) + rows, err := a.db.QueryxContext(ctx, "SELECT r.* FROM im_group_join_requests AS r WHERE "+strings.Join(where, " AND ")+" ORDER BY r.id LIMIT ?", args...) if err != nil { return nil, "", err } @@ -709,11 +763,6 @@ func (a *adapter) TenantGroupJoinRequestResolve(id t.TenantID, topic, requestID if err = updateGroupTopicSubCountTx(tx, topic); err != nil { return nil, err } - if err = insertGroupEventTx(tx, topic, actor, "group.join_request."+status, map[string]any{ - "request_id": requestID, "applicant_user_id": current.ApplicantUserID.UserId(), "status": status, - }); err != nil { - return nil, err - } if err = tx.Commit(); err != nil { return nil, err } @@ -791,11 +840,6 @@ func (a *adapter) TenantGroupAnnouncementPut(id t.TenantID, ann *t.GroupAnnounce VALUES(?,?,?,?,?,?,?,?,?,?)`, ann.AnnouncementID, ann.Topic, decodeUID(ann.AuthorUserID), ann.Content, t.GroupAnnouncementStatusCurrent, version, now, ann.IdempotencyKey, now, now); err != nil { return nil, err } - if err = insertGroupEventTx(tx, ann.Topic, ann.AuthorUserID, "group.announcement.updated", map[string]any{ - "announcement_id": ann.AnnouncementID, "version": version, - }); err != nil { - return nil, err - } if err = tx.Commit(); err != nil { return nil, err } @@ -897,11 +941,6 @@ func (a *adapter) TenantGroupMuteUpsert(id t.TenantID, mute *t.GroupMute) (*t.Gr if err != nil { return nil, err } - if err = insertGroupEventTx(tx, mute.Topic, mute.MutedBy, "group.mute.updated", map[string]any{ - "user_id": mute.UserID.UserId(), "muted": true, "expires_at": mute.ExpiresAt, - }); err != nil { - return nil, err - } if err = tx.Commit(); err != nil { return nil, err } @@ -931,11 +970,6 @@ func (a *adapter) TenantGroupMuteDelete(id t.TenantID, topic string, uid, actor if affected, _ := res.RowsAffected(); affected == 0 { return t.ErrNotFound } - if err = insertGroupEventTx(tx, topic, actor, "group.mute.updated", map[string]any{ - "user_id": uid.UserId(), "muted": false, - }); err != nil { - return err - } return tx.Commit() } @@ -1199,11 +1233,6 @@ func (a *adapter) TenantGroupCreate(id t.TenantID, topic *t.Topic, owner t.Uid, if err = updateGroupTopicSubCountTx(tx, topic.Id); err != nil { return err } - if err = insertGroupEventTx(tx, topic.Id, owner, "group.created", map[string]any{ - "owner_user_id": owner.UserId(), "member_count": len(seen), - }); err != nil { - return err - } return tx.Commit() } @@ -1261,11 +1290,6 @@ func (a *adapter) TenantGroupInviteMember(id t.TenantID, topic string, inviter, if err = updateGroupTopicSubCountTx(tx, topic); err != nil { return false, err } - if err = insertGroupEventTx(tx, topic, inviter, "group.member.joined", map[string]any{ - "user_id": target.UserId(), "source": "invitation", - }); err != nil { - return false, err - } if err = tx.Commit(); err != nil { return false, err } @@ -1304,9 +1328,6 @@ func (a *adapter) TenantGroupLeave(id t.TenantID, topic string, uid t.Uid, now t if err = updateGroupTopicSubCountTx(tx, topic); err != nil { return err } - if err = insertGroupEventTx(tx, topic, uid, "group.member.left", map[string]any{"user_id": uid.UserId()}); err != nil { - return err - } return tx.Commit() } @@ -1342,11 +1363,6 @@ func (a *adapter) TenantGroupRoleUpdate(id t.TenantID, topic string, actor, targ if affected, _ := res.RowsAffected(); affected == 0 { return t.ErrNotFound } - if err = insertGroupEventTx(tx, topic, actor, "group.role.updated", map[string]any{ - "user_id": target.UserId(), "role": role, - }); err != nil { - return err - } return tx.Commit() } @@ -1409,11 +1425,6 @@ func (a *adapter) TenantGroupOwnerTransfer(id t.TenantID, topic string, previous if affected, _ := res.RowsAffected(); affected == 0 { return t.ErrInvalidResponse } - if err = insertGroupEventTx(tx, topic, previousOwner, "group.owner.transferred", map[string]any{ - "previous_owner": previousOwner.UserId(), "owner": newOwner.UserId(), - }); err != nil { - return err - } return tx.Commit() } @@ -1441,49 +1452,6 @@ func (a *adapter) TenantGroupCanPost(id t.TenantID, topic string, uid t.Uid, adm return !muteAll && !memberMuted, nil } -func (a *adapter) TenantGroupEventListPending(id t.TenantID, limit int, now time.Time) ([]t.GroupEvent, error) { - if tenantAdp, err := a.routeGroupTenant(id); err != nil { - return nil, err - } else if tenantAdp != a { - return tenantAdp.TenantGroupEventListPending(id, limit, now) - } - ctx, cancel := a.getContext() - if cancel != nil { - defer cancel() - } - rows, err := a.db.QueryxContext(ctx, `SELECT * FROM im_group_events_outbox - WHERE status='pending' AND (next_retry_at IS NULL OR next_retry_at<=?) ORDER BY id LIMIT ?`, now, limit) - if err != nil { - return nil, err - } - defer rows.Close() - events := make([]t.GroupEvent, 0, limit) - for rows.Next() { - var event t.GroupEvent - if err = rows.StructScan(&event); err != nil { - return nil, err - } - event.ActorUserID = store.EncodeUid(event.ActorRaw) - events = append(events, event) - } - return events, rows.Err() -} - -func (a *adapter) TenantGroupEventMarkDelivered(id t.TenantID, eventID string, now time.Time) error { - if tenantAdp, err := a.routeGroupTenant(id); err != nil { - return err - } else if tenantAdp != a { - return tenantAdp.TenantGroupEventMarkDelivered(id, eventID, now) - } - ctx, cancel := a.getContext() - if cancel != nil { - defer cancel() - } - _, err := a.db.ExecContext(ctx, `UPDATE im_group_events_outbox SET status='delivered',attempts=attempts+1, - delivered_at=?,updated_at=? WHERE event_id=? AND status='pending'`, now, now, eventID) - return err -} - func insertGroupSettingsTx(tx *sqlx.Tx, settings *t.GroupSettings) error { settingsID, err := nextSnowflakeID() if err != nil { @@ -1523,12 +1491,6 @@ func updateGroupTopicSubCountTx(tx *sqlx.Tx, topic string) error { return err } -func insertGroupEventTx(tx *sqlx.Tx, topic string, actor t.Uid, eventType string, payload any) error { - _, err := tx.Exec(`INSERT INTO im_group_events_outbox(event_id,topic,actor_user_id,event_type,payload,status) - VALUES(?,?,?,?,?,'pending')`, "gev_"+store.Store.GetUidString(), topic, decodeUID(actor), eventType, common.ToJSON(payload)) - return err -} - func (a *adapter) routeGroupTenant(id t.TenantID) (*adapter, error) { if !a.tenantScopedDB { tenantAdp, err := a.routedTenantAdapter(id) @@ -1579,6 +1541,21 @@ func (a *adapter) groupJoinRequestByQuery(ctx context.Context, query string, arg return &item, nil } +func (a *adapter) groupInvitationByQuery(ctx context.Context, query string, args ...any) (*t.GroupInvitation, error) { + var item t.GroupInvitation + var handledAt sql.NullTime + err := a.db.QueryRowxContext(ctx, query, args...).Scan(&item.ID, &item.InvitationID, &item.Topic, &item.InviterUserRaw, &item.TargetUserRaw, + &item.Message, &item.Status, &item.Result, &item.JoinRequestID, &item.IdempotencyKey, &item.ExpiresAt, &handledAt, &item.CreatedAt, &item.UpdatedAt) + if err == sql.ErrNoRows { + return nil, t.ErrNotFound + } + if err != nil { + return nil, err + } + scanGroupInvitation(&item, handledAt) + return &item, nil +} + func (a *adapter) groupAnnouncementByQuery(ctx context.Context, query string, args ...any) (*t.GroupAnnouncement, error) { var item t.GroupAnnouncement err := a.db.QueryRowxContext(ctx, query, args...).StructScan(&item) @@ -1653,6 +1630,14 @@ func scanGroupJoinRequest(item *t.GroupJoinRequest) { } } +func scanGroupInvitation(item *t.GroupInvitation, handledAt sql.NullTime) { + item.InviterUserID = store.EncodeUid(item.InviterUserRaw) + item.TargetUserID = store.EncodeUid(item.TargetUserRaw) + if handledAt.Valid { + item.HandledAt = &handledAt.Time + } +} + func scanGroupMute(item *t.GroupMute) { item.UserID = store.EncodeUid(item.UserIDRaw) item.MutedBy = store.EncodeUid(item.MutedByRaw) diff --git a/server/im_http.go b/server/im_http.go index 88d3bcb..2bd0314 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -122,11 +122,6 @@ func withIMHTTPProtocol(prefix string, next http.Handler) http.Handler { return } req = req.WithContext(context.WithValue(req.Context(), imHTTPProtocolValidated, true)) - if req.Method == http.MethodGet || req.Method == http.MethodHead { - if principal, err := resolveIMHTTPBearer(req); err == nil { - dispatchIMBusinessEvents(principal.TenantID) - } - } serveIMHTTPIdempotent(wrt, req, next) }) } @@ -187,7 +182,6 @@ func serveIMHTTPIdempotent(wrt http.ResponseWriter, req *http.Request, next http next.ServeHTTP(wrt, req) return } - dispatchIMBusinessEvents(principal.TenantID) key := strings.TrimSpace(req.Header.Get("Idempotency-Key")) if key == "" || len(key) > 128 { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "Idempotency-Key is required", map[string]any{"field": "Idempotency-Key"}) @@ -514,6 +508,7 @@ type imGroupPatchRequest struct { type imGroupMemberSettingsRequest struct { Pinned *bool `json:"pinned,omitempty"` MuteNotifications *bool `json:"mute_notifications,omitempty"` + Version int `json:"version,omitempty"` } type imGroupMemberProfileUpdateRequest struct { @@ -570,12 +565,18 @@ type imGroupOwnerTransferRequest struct { Version int `json:"version,omitempty"` } +type imGroupDissolveRequest struct { + Reason string `json:"reason,omitempty"` +} + type imGroupMuteBatchRequest struct { Members []struct { UserID string `json:"user_id"` ExpiresAt string `json:"expires_at,omitempty"` } `json:"members"` - Reason string `json:"reason,omitempty"` + UserIDs []string `json:"user_ids,omitempty"` + Reason string `json:"reason,omitempty"` + Version int `json:"version,omitempty"` } type imSystemMessageActor struct { @@ -1248,6 +1249,7 @@ func serveIMPasswordResetConfirm(wrt http.ResponseWriter, req *http.Request) { if globals.sessionStore != nil { globals.sessionStore.EvictUser(vc.TargetUser, "") } + publishRealtimeEvictUser(tenant.ID, vc.TargetUser, "password_reset", "") writeIMSuccess(wrt, req, map[string]any{"status": "ok"}) } @@ -4320,10 +4322,7 @@ func serveIMMessageRecall(wrt http.ResponseWriter, req *http.Request) { } topic = normalizeConversationSettingsTopic(topic, principal.Uid) var body imMessageRecallRequest - if !decodeIMJSONBody(wrt, req, &body) || body.Version <= 0 { - if body.Version <= 0 { - writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "version"}) - } + if !decodeIMJSONBody(wrt, req, &body) { return } msg, authorized, err := authorizeIMMessageMutation(principal, topic, seq, true) @@ -4339,8 +4338,12 @@ func serveIMMessageRecall(wrt http.ResponseWriter, req *http.Request) { writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "message recall window expired", nil) return } + version := body.Version + if version <= 0 { + version = msg.Version + } recalledAt := time.Now().UTC().Round(time.Millisecond) - updated, err := store.Messages.Recall(principal.TenantID, topic, seq, principal.Uid, body.Version, recalledAt) + updated, err := store.Messages.Recall(principal.TenantID, topic, seq, principal.Uid, version, recalledAt) if err != nil { writeIMMessageMutationError(wrt, req, err) return @@ -4683,7 +4686,6 @@ func serveIMFriendRequestCreate(wrt http.ResponseWriter, req *http.Request) { writeIMRelationError(wrt, req, err) return } - dispatchIMBusinessEvents(principal.TenantID) if imReliableHTTPEventsEnabled(req) { if err = publishIMRelationEvent(principal.TenantID, principal.Uid, target, created.RequestID, "friend.request.created", "新的好友申请", message); err != nil { @@ -4864,7 +4866,8 @@ func serveIMGroupCreate(wrt http.ResponseWriter, req *http.Request) { writeIMRelationError(wrt, req, err) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topicName, principal.Uid, "group.created", + map[string]any{"topic": topicName, "member_user_ids": userIDs(memberIDs)}) created, err := store.Topics.Get(principal.TenantID, topicName) if err != nil { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) @@ -4888,7 +4891,7 @@ func serveIMGroupResource(wrt http.ResponseWriter, req *http.Request) { serveIMGroupSettings(wrt, req, topic) return } - if tail == "member-settings" { + if tail == "member-settings" || tail == "members/me/settings" { serveIMGroupMemberSettings(wrt, req, topic) return } @@ -5098,6 +5101,9 @@ func serveIMGroupDetail(wrt http.ResponseWriter, req *http.Request, topic string } public := groupPublicMap(group.Public) currentVersion := intFromAny(public["version"]) + if settings, err := store.GroupBusiness.GetSettings(principal.TenantID, topic); err == nil && settings != nil && settings.Version > 0 { + currentVersion = settings.Version + } if currentVersion <= 0 { currentVersion = 1 } @@ -5151,7 +5157,8 @@ func serveIMGroupDetail(wrt http.ResponseWriter, req *http.Request, topic string return } } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.settings.updated", + map[string]any{"topic": topic}) updated, err := store.Topics.Get(principal.TenantID, topic) if err != nil { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) @@ -5164,6 +5171,33 @@ func serveIMGroupDetail(wrt http.ResponseWriter, req *http.Request, topic string } } writeIMSuccess(wrt, req, payload) + case http.MethodDelete: + if !(selfSub.ModeGiven & selfSub.ModeWant).IsOwner() { + writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil) + return + } + if !consumeIMActionTokenHeader(wrt, req, principal, "group_dissolve") { + return + } + var body imGroupDissolveRequest + if req.Body != nil && req.Body != http.NoBody { + if !decodeIMJSONBody(wrt, req, &body) { + return + } + } + settings, err := store.GroupBusiness.Dissolve(principal.TenantID, topic, principal.Uid, strings.TrimSpace(body.Reason), time.Now().UTC().Round(time.Millisecond)) + if err != nil { + writeIMRelationError(wrt, req, err) + return + } + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.dissolved", + map[string]any{"topic": topic}) + payload := map[string]any{"topic": topic, "deleted": true} + if settings != nil { + payload["status"] = settings.GroupStatus + payload["version"] = settings.Version + } + writeIMSuccess(wrt, req, payload) default: writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) } @@ -5205,13 +5239,21 @@ func serveIMGroupMemberSettings(wrt http.ResponseWriter, req *http.Request, topi return } private := groupPrivateMap(selfSub.Private) + currentVersion := intFromAny(private["version"]) + if currentVersion <= 0 { + currentVersion = 1 + } + if body.Version > 0 && body.Version != currentVersion { + writeIMError(wrt, req, http.StatusConflict, 30006, "VERSION_CONFLICT", "version conflict", map[string]any{"current_version": currentVersion}) + return + } if body.Pinned != nil { private["pinned"] = *body.Pinned } if body.MuteNotifications != nil { private["mute_notifications"] = *body.MuteNotifications } - private["version"] = intFromAny(private["version"]) + 1 + private["version"] = currentVersion + 1 private["updated_at"] = time.Now().UTC().Round(time.Millisecond).Format(time.RFC3339Nano) if err := store.Subs.Update(principal.TenantID, topic, principal.Uid, map[string]any{"Private": private}); err != nil { writeIMRelationError(wrt, req, err) @@ -5224,7 +5266,8 @@ func serveIMGroupMemberSettings(wrt http.ResponseWriter, req *http.Request, topi writeIMRelationError(wrt, req, err) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.member.updated", + map[string]any{"user_id": principal.Uid.UserId()}) selfSub.Private = private writeIMSuccess(wrt, req, groupMemberSettingsPayload(selfSub)) } @@ -5275,7 +5318,8 @@ func serveIMGroupMyProfile(wrt http.ResponseWriter, req *http.Request, topic str if _, err := store.GroupBusiness.UpdateMember(principal.TenantID, topic, principal.Uid, types.GroupMemberPatch{Nickname: &nickname}); err != nil { if errors.Is(err, types.ErrNotFound) { logs.Warn.Printf("group[%s]: member profile saved to subscription only; business member row missing for user %s", topic, principal.Uid.UserId()) - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.member.updated", + map[string]any{"user_id": principal.Uid.UserId()}) writeIMSuccess(wrt, req, map[string]any{"nickname": nickname, "version": currentVersion + 1, "updated_at": updatedAt.Format(time.RFC3339Nano)}) return } @@ -5285,7 +5329,8 @@ func serveIMGroupMyProfile(wrt http.ResponseWriter, req *http.Request, topic str writeIMRelationError(wrt, req, err) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.member.updated", + map[string]any{"user_id": principal.Uid.UserId()}) writeIMSuccess(wrt, req, map[string]any{"nickname": nickname, "version": currentVersion + 1, "updated_at": updatedAt.Format(time.RFC3339Nano)}) } @@ -5306,7 +5351,8 @@ func serveIMGroupLeave(wrt http.ResponseWriter, req *http.Request, topic string) writeIMRelationError(wrt, req, err) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.member.left", + map[string]any{"user_id": principal.Uid.UserId()}) writeIMSuccess(wrt, req, map[string]any{"topic": topic, "status": "left"}) } @@ -5466,7 +5512,8 @@ func serveIMGroupInvitations(wrt http.ResponseWriter, req *http.Request, topic s publishIMGroupInvitationNotification(principal.TenantID, inv) results = append(results, item) } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.invitations.updated", + map[string]any{"topic": topic, "items": results}) writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": results}) } @@ -5630,7 +5677,8 @@ func serveIMGroupInvitationResource(wrt http.ResponseWriter, req *http.Request) []imSystemMessageAction{{Type: "group_profile", Label: "查看群聊", TargetID: inv.Topic}}, map[string]any{"invitation_id": inv.InvitationID, "source": "invitation"}) } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, inv.Topic, principal.Uid, "group.member.joined", + map[string]any{"user_id": principal.Uid.UserId(), "invitation_id": inv.InvitationID}) writeIMSuccess(wrt, req, groupInvitationPayload(inv)) } @@ -5665,7 +5713,8 @@ func serveIMGroupJoinRequests(wrt http.ResponseWriter, req *http.Request, topic writeIMRelationError(wrt, req, joinErr) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.member.joined", + map[string]any{"user_id": principal.Uid.UserId(), "joined": joined, "source": "open_group"}) writeIMSuccess(wrt, req, map[string]any{"topic": topic, "status": "joined", "joined": joined, "source": "open_group"}) return } @@ -5694,7 +5743,8 @@ func serveIMGroupJoinRequests(wrt http.ResponseWriter, req *http.Request, topic writeIMRelationError(wrt, req, err) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.join_request.created", + map[string]any{"request_id": item.RequestID, "user_id": principal.Uid.UserId()}) writeIMSuccess(wrt, req, groupJoinRequestPayload(item)) return } @@ -5723,6 +5773,7 @@ func serveIMGroupJoinRequests(wrt http.ResponseWriter, req *http.Request, topic } items = nil } + items = dedupePendingGroupJoinRequests(items, status) payloads, err := groupJoinRequestPayloadsWithUsers(principal.TenantID, items) if err != nil { writeIMRelationError(wrt, req, err) @@ -5771,7 +5822,8 @@ func serveIMGroupJoinRequestAction(wrt http.ResponseWriter, req *http.Request, t []imSystemMessageAction{{Type: "group_profile", Label: "查看群聊", TargetID: topic}}, map[string]any{"request_id": updated.RequestID, "source": "approval"}) } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.join_request."+updated.Status, + map[string]any{"request_id": updated.RequestID, "user_id": updated.ApplicantUserID.UserId(), "status": updated.Status}) writeIMSuccess(wrt, req, groupJoinRequestPayload(updated)) } @@ -5843,7 +5895,8 @@ func serveIMGroupAnnouncements(wrt http.ResponseWriter, req *http.Request, topic writeIMRelationError(wrt, req, err) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.announcement.updated", + map[string]any{"announcement_id": announcement.AnnouncementID, "topic": topic}) writeIMSuccess(wrt, req, groupAnnouncementPayloadWithPermissions(topic, announcement, selfSub)) case strings.HasPrefix(tail, "announcements/") && strings.HasSuffix(tail, "/read") && req.Method == http.MethodPost: announcementID := strings.TrimSuffix(strings.TrimPrefix(tail, "announcements/"), "/read") @@ -6001,7 +6054,8 @@ func serveIMGroupGovernance(wrt http.ResponseWriter, req *http.Request, topic st nil, map[string]any{"mute_all": *body.MuteAll, "version": settings.Version}) } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.settings.updated", + map[string]any{"topic": topic, "version": settings.Version}) writeIMSuccess(wrt, req, groupGovernancePayloadFromSettings(settings, selfSub)) default: writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) @@ -6082,7 +6136,8 @@ func serveIMGroupAdmins(wrt http.ResponseWriter, req *http.Request, topic, tail writeIMRelationError(wrt, req, err) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.role.updated", + map[string]any{"topic": topic, "results": results}) writeIMSuccess(wrt, req, map[string]any{"succeeded": results, "admins": groupAdminItems(members), "version": body.Version + 1, "updated_at": time.Now().UTC().Format(time.RFC3339Nano)}) case strings.HasPrefix(tail, "admins/") && req.Method == http.MethodDelete: userID := strings.TrimPrefix(tail, "admins/") @@ -6096,7 +6151,8 @@ func serveIMGroupAdmins(wrt http.ResponseWriter, req *http.Request, topic, tail writeIMRelationError(wrt, req, err) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.role.updated", + map[string]any{"user_id": uid.UserId(), "role": "member"}) writeIMSuccess(wrt, req, map[string]any{"user_id": uid.UserId(), "role": "member", "updated_at": time.Now().UTC().Format(time.RFC3339Nano)}) default: writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) @@ -6120,6 +6176,9 @@ func serveIMGroupOwnerTransfer(wrt http.ResponseWriter, req *http.Request, topic if !decodeIMJSONBody(wrt, req, &body) { return } + if !consumeIMActionTokenHeader(wrt, req, principal, "group_owner_transfer") { + return + } target := types.ParseUserId(strings.TrimSpace(body.TargetUserID)) if target.IsZero() || target == principal.Uid { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "target_user_id"}) @@ -6135,7 +6194,8 @@ func serveIMGroupOwnerTransfer(wrt http.ResponseWriter, req *http.Request, topic writeIMRelationError(wrt, req, err) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.owner.transferred", + map[string]any{"owner": target.UserId(), "previous_owner": principal.Uid.UserId()}) writeIMSuccess(wrt, req, map[string]any{"owner": target.UserId(), "previous_owner": principal.Uid.UserId(), "version": body.Version + 1}) } @@ -6162,8 +6222,16 @@ func serveIMGroupMutes(wrt http.ResponseWriter, req *http.Request, topic, tail s return } if len(body.Members) == 0 || len(body.Members) > 100 { - writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "members"}) - return + for _, userID := range cleanStringList(body.UserIDs, 100) { + body.Members = append(body.Members, struct { + UserID string `json:"user_id"` + ExpiresAt string `json:"expires_at,omitempty"` + }{UserID: userID}) + } + if len(body.Members) == 0 || len(body.Members) > 100 { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "members"}) + return + } } now := time.Now().UTC().Round(time.Millisecond) mutes := make([]types.GroupMute, 0, len(body.Members)) @@ -6185,7 +6253,8 @@ func serveIMGroupMutes(wrt http.ResponseWriter, req *http.Request, topic, tail s } mutes = append(mutes, *mute) } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.mute.updated", + map[string]any{"topic": topic, "items": groupMutePayloads(mutes)}) writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": groupMutePayloads(mutes)}) case strings.HasPrefix(tail, "mutes/") && req.Method == http.MethodDelete: uid := types.ParseUserId(strings.TrimSpace(strings.TrimPrefix(tail, "mutes/"))) @@ -6197,7 +6266,8 @@ func serveIMGroupMutes(wrt http.ResponseWriter, req *http.Request, topic, tail s writeIMRelationError(wrt, req, err) return } - dispatchIMGroupEvents(principal.TenantID) + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.mute.updated", + map[string]any{"user_id": uid.UserId(), "muted": false}) writeIMSuccess(wrt, req, map[string]any{"user_id": uid.UserId(), "muted": false}) default: writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) @@ -6639,7 +6709,6 @@ func serveIMFriendRequestAction(wrt http.ResponseWriter, req *http.Request) { {UserID: updated.TargetID.UserId(), Click: map[string]any{"type": "user_profile", "target_id": updated.TargetID.UserId()}}, }, nil, map[string]any{"request_id": updated.RequestID}) } - dispatchIMBusinessEvents(principal.TenantID) if imReliableHTTPEventsEnabled(req) { eventType := "friend.request." + status if err = publishIMRelationEvent(principal.TenantID, principal.Uid, updated.ApplicantID, @@ -7163,11 +7232,30 @@ func defaultIMActionTarget(action, fallback string) string { switch action { case "set_independent_password", "enable_youth_mode", "close_youth_mode", "reset_independent_password": return "youth_mode" + case "group_owner_transfer", "group_dissolve": + return action default: return fallback } } +func consumeIMActionTokenHeader(wrt http.ResponseWriter, req *http.Request, principal *imPrincipal, action string) bool { + actionToken := strings.TrimSpace(req.Header.Get("X-Action-Token")) + if actionToken == "" { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "X-Action-Token"}) + return false + } + if store.Security == nil { + writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) + return false + } + if _, err := store.Security.ConsumeActionToken(principal.TenantID, principal.Uid, actionToken, action, defaultIMActionTarget(action, action), time.Now().UTC().Round(time.Millisecond)); err != nil { + writeIMError(wrt, req, http.StatusForbidden, 30042, imReasonActionToken, "action token invalid", nil) + return false + } + return true +} + func hashIMActionToken(tenantID types.TenantID, uid types.Uid, action, target, tokenID, secret string) string { h := sha256.New() h.Write([]byte(fmt.Sprintf("%d", tenantID))) @@ -7930,6 +8018,26 @@ func groupJoinRequestPayloads(items []types.GroupJoinRequest) []map[string]any { return out } +func dedupePendingGroupJoinRequests(items []types.GroupJoinRequest, status string) []types.GroupJoinRequest { + if len(items) < 2 || status != types.GroupJoinRequestStatusPending { + return items + } + seen := make(map[string]struct{}, len(items)) + deduped := make([]types.GroupJoinRequest, 0, len(items)) + for i := len(items) - 1; i >= 0; i-- { + key := items[i].Topic + ":" + items[i].ApplicantUserID.UserId() + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + deduped = append(deduped, items[i]) + } + for left, right := 0, len(deduped)-1; left < right; left, right = left+1, right-1 { + deduped[left], deduped[right] = deduped[right], deduped[left] + } + return deduped +} + func groupJoinRequestPayloadsWithUsers(tenantID types.TenantID, items []types.GroupJoinRequest) ([]map[string]any, error) { payloads := groupJoinRequestPayloads(items) if len(items) == 0 || store.Users == nil { @@ -9327,7 +9435,8 @@ func userProfilePayload10(profile *types.UserProfile) map[string]any { func validIMSecurityChallengeScene(scene string) bool { switch strings.TrimSpace(scene) { - case "set_independent_password", "enable_youth_mode", "close_youth_mode", "reset_independent_password": + case "set_independent_password", "enable_youth_mode", "close_youth_mode", "reset_independent_password", + "group_owner_transfer", "group_dissolve": return true default: return false @@ -9930,7 +10039,6 @@ func requireIMGroupMember(wrt http.ResponseWriter, req *http.Request, topic stri writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"}) return nil, nil, false } - dispatchIMGroupEvents(principal.TenantID) sub, err := store.Subs.Get(principal.TenantID, topic, principal.Uid, false) if err != nil { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) @@ -11297,6 +11405,7 @@ func broadcastIMConversationSystemMessage(tenantID types.TenantID, msg *types.Me logs.Warn.Printf("im system message: hub route queue full topic=%s seq=%d", msg.Topic, msg.SeqId) } deliverIMConversationSystemMessageToP2PSessions(tenantID, msg, out) + publishRealtimeTopicMessage(msg.Topic, out) broadcastIMConversationSystemPresence(tenantID, msg, timestamp) } diff --git a/server/im_http_test.go b/server/im_http_test.go index c898d1d..1719290 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -41,7 +41,6 @@ type testVerificationStore struct { type testHTTPInfrastructureStore struct { records map[string]*types.HTTPIdempotencyRecord - events []types.BusinessEvent deadLetters []types.KafkaDeadLetter } @@ -137,19 +136,6 @@ func (s *testHTTPInfrastructureStore) AbortIdempotency(_ types.TenantID, record return nil } -func (s *testHTTPInfrastructureStore) CreateEvent(_ types.TenantID, event *types.BusinessEvent) error { - s.events = append(s.events, *event) - return nil -} - -func (s *testHTTPInfrastructureStore) ListPendingEvents(_ types.TenantID, _ int, _ time.Time) ([]types.BusinessEvent, error) { - return append([]types.BusinessEvent(nil), s.events...), nil -} - -func (s *testHTTPInfrastructureStore) MarkEventDelivered(_ types.TenantID, _ string, _ time.Time) error { - return nil -} - func (s *testHTTPInfrastructureStore) BeginKafkaConsume(_ types.TenantID, record *types.KafkaConsumeRecord, _ time.Time) (*types.KafkaConsumeRecord, bool, error) { copy := *record copy.Status = types.KafkaConsumeProcessing @@ -403,6 +389,16 @@ func (s *testGroupBusinessStore) UpdateSettings(_ types.TenantID, topic string, return settings, nil } +func (s *testGroupBusinessStore) Dissolve(_ types.TenantID, topic string, actor types.Uid, reason string, now time.Time) (*types.GroupSettings, error) { + settings, _ := s.GetSettings(0, topic) + settings.GroupStatus = "dissolved" + settings.DissolvedBy = actor + settings.DissolvedAt = &now + settings.Version++ + s.settings = settings + return settings, nil +} + func (s *testGroupBusinessStore) UpsertMember(_ types.TenantID, member *types.GroupMember) (*types.GroupMember, error) { s.member = member return member, nil @@ -426,6 +422,20 @@ func (s *testGroupBusinessStore) UpdateMember(_ types.TenantID, topic string, ui } func (s *testGroupBusinessStore) CreateInvitation(_ types.TenantID, inv *types.GroupInvitation) (*types.GroupInvitation, error) { + if inv.JoinRequestID != "" { + for i := len(s.invitations) - 1; i >= 0; i-- { + existing := &s.invitations[i] + if existing.Topic == inv.Topic && existing.TargetUserID == inv.TargetUserID && + existing.Status == types.GroupInvitationStatusPending && existing.JoinRequestID != "" { + for j := len(s.joinRequests) - 1; j >= 0; j-- { + request := &s.joinRequests[j] + if request.RequestID == existing.JoinRequestID && request.Status == types.GroupJoinRequestStatusPending { + return existing, nil + } + } + } + } + } s.invitations = append(s.invitations, *inv) if inv.JoinRequestID != "" { expiresAt := inv.ExpiresAt @@ -464,6 +474,13 @@ func (s *testGroupBusinessStore) AcceptInvitation(_ types.TenantID, invitationID } func (s *testGroupBusinessStore) CreateJoinRequest(_ types.TenantID, req *types.GroupJoinRequest) (*types.GroupJoinRequest, error) { + for i := len(s.joinRequests) - 1; i >= 0; i-- { + existing := &s.joinRequests[i] + if existing.Topic == req.Topic && existing.ApplicantUserID == req.ApplicantUserID && + existing.Status == types.GroupJoinRequestStatusPending { + return existing, nil + } + } now := time.Now().UTC() req.CreatedAt = now req.UpdatedAt = now @@ -613,14 +630,6 @@ func (s *testGroupBusinessStore) CanPost(_ types.TenantID, topic string, uid typ return true, nil } -func (s *testGroupBusinessStore) ListPendingEvents(_ types.TenantID, _ int, _ time.Time) ([]types.GroupEvent, error) { - return nil, nil -} - -func (s *testGroupBusinessStore) MarkEventDelivered(_ types.TenantID, _ string, _ time.Time) error { - return nil -} - func defaultStringPtr(value *string, fallback string) string { if value == nil { return fallback @@ -2529,6 +2538,60 @@ func TestServeIMSecurityChallengeVerifyIssuesActionToken(t *testing.T) { } } +func TestServeIMSecurityChallengeSupportsGroupSensitiveScenes(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + expectIMBearer(t, ctrl, "token-1", 7, 42) + oldUsers := store.Users + oldVerifications := store.Verifications + oldSecurity := store.Security + oldSender := sendIMVerificationCode + users := mock_store.NewMockUsersPersistenceInterface(ctrl) + users.EXPECT().Get(types.TenantID(7), types.Uid(42)).Return(&types.User{ + ObjHeader: types.ObjHeader{Id: types.Uid(42).String()}, + TenantID: 7, + Phone: "+8613800138000", + }, nil) + verifications := &testVerificationStore{} + security := &testSecurityStore{} + store.Users = users + store.Verifications = verifications + store.Security = security + sendIMVerificationCode = func(_ types.TenantID, _, _, _, _ string) error { return nil } + defer func() { + store.Users = oldUsers + store.Verifications = oldVerifications + store.Security = oldSecurity + sendIMVerificationCode = oldSender + }() + + req := httptest.NewRequest(http.MethodPost, "/v1/im/security/challenges", + bytes.NewBufferString(`{"scene":"group_owner_transfer","channel":"sms"}`)) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + serveIMSecurityChallenges(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("create status: got %d body=%s", rec.Code, rec.Body.String()) + } + if verifications.created == nil || verifications.created.Action != "group_owner_transfer" { + t.Fatalf("unexpected created verification: %#v", verifications.created) + } + verifications.get = verifications.created + + expectIMBearer(t, ctrl, "token-1", 7, 42) + verifyReq := httptest.NewRequest(http.MethodPost, "/v1/im/security/challenges/"+verifications.created.ChallengeID+"/verify", + bytes.NewBufferString(`{"code":"123456"}`)) + verifyReq.Header.Set("Authorization", "Bearer token-1") + verifyRec := httptest.NewRecorder() + serveIMSecurityChallengeVerify(verifyRec, verifyReq) + if verifyRec.Code != http.StatusOK { + t.Fatalf("verify status: got %d body=%s", verifyRec.Code, verifyRec.Body.String()) + } + if security.actionToken == nil || security.actionToken.Action != "group_owner_transfer" || security.actionToken.Target != "group_owner_transfer" { + t.Fatalf("unexpected group action token: %#v", security.actionToken) + } +} + func TestServeIMSecurityChallengeCreateEnforcesResendLimit(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -2955,11 +3018,9 @@ func TestServeIMMeetingsCreate(t *testing.T) { oldKafkaBus := kafkaBus oldPublisher := publishKafkaAuditEventEnvelope oldSecurity := store.Security - oldInfrastructure := store.HTTPInfrastructure oldMeetings := store.Meetings meetings := &testMeetingsStore{} security := &testSecurityStore{} - infrastructure := &testHTTPInfrastructureStore{} var auditPublished bool kafkaBus = &kafkaRuntime{} publishKafkaAuditEventEnvelope = func(_ context.Context, topicName, key string, envelope kafkaEnvelope) error { @@ -2968,13 +3029,11 @@ func TestServeIMMeetingsCreate(t *testing.T) { return nil } store.Security = security - store.HTTPInfrastructure = infrastructure store.Meetings = meetings defer func() { kafkaBus = oldKafkaBus publishKafkaAuditEventEnvelope = oldPublisher store.Security = oldSecurity - store.HTTPInfrastructure = oldInfrastructure store.Meetings = oldMeetings }() @@ -2998,8 +3057,8 @@ func TestServeIMMeetingsCreate(t *testing.T) { if !auditPublished { t.Fatalf("meeting create should still publish audit event") } - if len(security.notifications) != 0 || len(infrastructure.events) != 0 { - t.Fatalf("meeting create HTTP layer must not duplicate workbench notifications/events: notifications=%#v events=%#v", security.notifications, infrastructure.events) + if len(security.notifications) != 0 { + t.Fatalf("meeting create HTTP layer must not duplicate workbench notifications: notifications=%#v", security.notifications) } } @@ -4542,7 +4601,7 @@ func TestServeIMGroupDetailPatchSettingsProfileAndLeave(t *testing.T) { t.Fatalf("unexpected patch response: %#v", data) } - settingsReq := httptest.NewRequest(http.MethodPatch, "/v1/im/groups/grpAlpha/member-settings", bytes.NewBufferString(`{"pinned":true,"mute_notifications":true}`)) + settingsReq := httptest.NewRequest(http.MethodPatch, "/v1/im/groups/grpAlpha/members/me/settings", bytes.NewBufferString(`{"pinned":true,"mute_notifications":true,"version":1}`)) settingsReq.Header.Set("Authorization", "Bearer token-1") settingsRec := httptest.NewRecorder() serveIMGroupResource(settingsRec, settingsReq) @@ -4649,6 +4708,66 @@ func TestServeIMGroupDetailPatchKeepsTopicProfileWhenBusinessSettingsMissing(t * } } +func TestServeIMGroupDetailPatchUsesBusinessSettingsVersion(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + oldStore := store.Store + oldGroupBusiness := store.GroupBusiness + mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) + tokenAuth := mock_auth.NewMockAuthHandler(ctrl) + mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth) + tokenAuth.EXPECT().IsInitialized().Return(true) + tokenAuth.EXPECT().Authenticate(auth.AuthContext{}, []byte("token-1")). + Return(&auth.Rec{TenantID: 7, Uid: 42, AuthLevel: auth.LevelAuth}, nil, nil) + store.Store = mockStore + oldTopics := store.Topics + oldSubs := store.Subs + topics := mock_store.NewMockTopicsPersistenceInterface(ctrl) + subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + store.Topics = topics + store.Subs = subs + groupStore := &testGroupBusinessStore{settings: &types.GroupSettings{ + Topic: "grpAlpha", OwnerUserID: types.Uid(42), Name: "Alpha", + GroupStatus: types.GroupStatusActive, Version: 7, + }} + store.GroupBusiness = groupStore + defer func() { + store.Store = oldStore + store.Topics = oldTopics + store.Subs = oldSubs + store.GroupBusiness = oldGroupBusiness + }() + selfSub := &types.Subscription{ + TenantID: 7, User: types.Uid(42).String(), Topic: "grpAlpha", + ModeWant: types.ModeCFull, ModeGiven: types.ModeCFull, + } + group := &types.Topic{ + ObjHeader: types.ObjHeader{Id: "grpAlpha"}, TenantID: 7, + Owner: types.Uid(42).String(), Public: map[string]any{"name": "Alpha", "version": 1}, + } + subs.EXPECT().Get(types.TenantID(7), "grpAlpha", types.Uid(42), false).Return(selfSub, nil) + topics.EXPECT().Get(types.TenantID(7), "grpAlpha").Return(group, nil) + topics.EXPECT().Update(types.TenantID(7), "grpAlpha", gomock.Any()).Return(nil) + topics.EXPECT().Get(types.TenantID(7), "grpAlpha").Return(&types.Topic{ + ObjHeader: types.ObjHeader{Id: "grpAlpha"}, TenantID: 7, + Owner: types.Uid(42).String(), Public: map[string]any{"name": "Beta", "version": 8}, + }, nil) + + req := httptest.NewRequest(http.MethodPatch, "/v1/im/groups/grpAlpha", bytes.NewBufferString(`{"name":"Beta","version":7}`)) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + serveIMGroupResource(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("patch status: got %d body=%s", rec.Code, rec.Body.String()) + } + if groupStore.updatedSettings.Version != 7 { + t.Fatalf("business settings update used version %d, want 7", groupStore.updatedSettings.Version) + } + if data := decodeIMHTTPResponse(t, rec).Data.(map[string]any); data["version"].(float64) != 8 { + t.Fatalf("unexpected patch response: %#v", data) + } +} + func TestServeIMGroupMyProfileKeepsSubscriptionNicknameWhenBusinessMemberMissing(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -4926,6 +5045,26 @@ func TestServeIMGroupInvitationCreatesApprovalRequest(t *testing.T) { } } +func TestDedupePendingGroupJoinRequestsKeepsLatestPerApplicant(t *testing.T) { + applicant := types.Uid(43) + other := types.Uid(44) + items := []types.GroupJoinRequest{ + {ID: 10, RequestID: "gjr_old", Topic: "grpAlpha", ApplicantUserID: applicant, Status: types.GroupJoinRequestStatusPending}, + {ID: 11, RequestID: "gjr_other", Topic: "grpAlpha", ApplicantUserID: other, Status: types.GroupJoinRequestStatusPending}, + {ID: 12, RequestID: "gjr_new", Topic: "grpAlpha", ApplicantUserID: applicant, Status: types.GroupJoinRequestStatusPending}, + } + + deduped := dedupePendingGroupJoinRequests(items, types.GroupJoinRequestStatusPending) + if len(deduped) != 2 || deduped[0].RequestID != "gjr_other" || deduped[1].RequestID != "gjr_new" { + t.Fatalf("unexpected deduped requests: %#v", deduped) + } + + all := dedupePendingGroupJoinRequests(items, "all") + if len(all) != len(items) { + t.Fatalf("non-pending list should not be deduped: %#v", all) + } +} + func TestServeIMGroupInvitationRespectsTargetPrivacyApproval(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -5415,6 +5554,7 @@ func TestServeIMGroupAdminsMutesAndOwnerTransfer(t *testing.T) { defer ctrl.Finish() oldStore := store.Store oldGroupBusiness := store.GroupBusiness + oldSecurity := store.Security mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) tokenAuth := mock_auth.NewMockAuthHandler(ctrl) mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth).Times(3) @@ -5430,11 +5570,14 @@ func TestServeIMGroupAdminsMutesAndOwnerTransfer(t *testing.T) { store.Subs = subs groupStore := &testGroupBusinessStore{} store.GroupBusiness = groupStore + security := &testSecurityStore{} + store.Security = security defer func() { store.Store = oldStore store.Topics = oldTopics store.Subs = oldSubs store.GroupBusiness = oldGroupBusiness + store.Security = oldSecurity }() owner := types.Uid(42) target := types.Uid(43) @@ -5456,7 +5599,7 @@ func TestServeIMGroupAdminsMutesAndOwnerTransfer(t *testing.T) { t.Fatalf("admin role mode should grant admin without owner: %s", groupStore.updatedRoleMode.String()) } - muteReq := httptest.NewRequest(http.MethodPost, "/v1/im/groups/grpAlpha/mutes", bytes.NewBufferString(`{"members":[{"user_id":"`+target.UserId()+`"}],"reason":"test"}`)) + muteReq := httptest.NewRequest(http.MethodPost, "/v1/im/groups/grpAlpha/mutes/batch", bytes.NewBufferString(`{"user_ids":["`+target.UserId()+`"],"reason":"test","version":1}`)) muteReq.Header.Set("Authorization", "Bearer token-1") muteRec := httptest.NewRecorder() serveIMGroupResource(muteRec, muteReq) @@ -5466,6 +5609,7 @@ func TestServeIMGroupAdminsMutesAndOwnerTransfer(t *testing.T) { transferReq := httptest.NewRequest(http.MethodPost, "/v1/im/groups/grpAlpha/owner-transfer", bytes.NewBufferString(`{"target_user_id":"`+target.UserId()+`","version":1}`)) transferReq.Header.Set("Authorization", "Bearer token-1") + transferReq.Header.Set("X-Action-Token", "action.1") transferRec := httptest.NewRecorder() serveIMGroupResource(transferRec, transferReq) if transferRec.Code != http.StatusOK { @@ -5474,6 +5618,57 @@ func TestServeIMGroupAdminsMutesAndOwnerTransfer(t *testing.T) { if data := decodeIMHTTPResponse(t, transferRec).Data.(map[string]any); data["owner"] != target.UserId() { t.Fatalf("unexpected transfer response: %#v", data) } + if security.consumedAction != "group_owner_transfer" || security.consumedTarget != "group_owner_transfer" { + t.Fatalf("owner transfer consumed wrong action token scope: action=%q target=%q", security.consumedAction, security.consumedTarget) + } +} + +func TestServeIMGroupDissolveConsumesActionToken(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + oldStore := store.Store + oldGroupBusiness := store.GroupBusiness + oldSecurity := store.Security + mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) + tokenAuth := mock_auth.NewMockAuthHandler(ctrl) + mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth) + tokenAuth.EXPECT().IsInitialized().Return(true) + tokenAuth.EXPECT().Authenticate(auth.AuthContext{}, []byte("token-1")). + Return(&auth.Rec{TenantID: 7, Uid: 42, AuthLevel: auth.LevelAuth}, nil, nil) + store.Store = mockStore + oldSubs := store.Subs + subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + store.Subs = subs + groupStore := &testGroupBusinessStore{} + store.GroupBusiness = groupStore + security := &testSecurityStore{} + store.Security = security + defer func() { + store.Store = oldStore + store.Subs = oldSubs + store.GroupBusiness = oldGroupBusiness + store.Security = oldSecurity + }() + ownerSub := &types.Subscription{ + TenantID: 7, User: types.Uid(42).String(), Topic: "grpAlpha", + ModeWant: types.ModeCFull, ModeGiven: types.ModeCFull, + } + subs.EXPECT().Get(types.TenantID(7), "grpAlpha", types.Uid(42), false).Return(ownerSub, nil) + + req := httptest.NewRequest(http.MethodDelete, "/v1/im/groups/grpAlpha", bytes.NewBufferString(`{"reason":"结束项目"}`)) + req.Header.Set("Authorization", "Bearer token-1") + req.Header.Set("X-Action-Token", "action.1") + rec := httptest.NewRecorder() + serveIMGroupResource(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("dissolve status: got %d body=%s", rec.Code, rec.Body.String()) + } + if security.consumedAction != "group_dissolve" || security.consumedTarget != "group_dissolve" { + t.Fatalf("dissolve consumed wrong action token scope: action=%q target=%q", security.consumedAction, security.consumedTarget) + } + if groupStore.settings == nil || groupStore.settings.GroupStatus != "dissolved" { + t.Fatalf("group not dissolved: %#v", groupStore.settings) + } } func TestServeIMFavoritesListDetailBatchDelete(t *testing.T) { @@ -7435,7 +7630,7 @@ func TestServeIMMessageRecall(t *testing.T) { ObjHeader: types.ObjHeader{UpdatedAt: time.Now().UTC()}, }, nil) - req := httptest.NewRequest(http.MethodPost, "/v1/im/topics/grpRecall/messages/9/recall", bytes.NewBufferString(`{"version":1}`)) + req := httptest.NewRequest(http.MethodPost, "/v1/im/topics/grpRecall/messages/9/recall", bytes.NewBufferString(`{}`)) req.Header.Set("Authorization", "Bearer token-1") rec := httptest.NewRecorder()