pro/im_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: repair group management edge cases

Browse Source
This commit is contained in:
richjohnrambo
2026-09-13 16:04:31 +08:00
parent a9afc6ff6a
commit 7b304412b0
3 changed files with 473 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/tenant_groups.go
+100 -115
View File
@@ -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)