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

fix tenant sharding for group topics

Browse Source
This commit is contained in:
richjohnrambo
2026-09-23 09:49:14 +08:00
parent a968fd252e
commit 6fa3009987
4 changed files with 31 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/tenant_groups.go
+20 -20
View File
@@ -17,7 +17,7 @@ import (
)
const (
groupSettingsSelectColumns = `topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,
groupSettingsSelectColumns = `tenant_id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,
member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,
created_by,dissolved_by,dissolved_at,created_at,updated_at`
groupAnnouncementSelectColumnsBase = `id,announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at,
@@ -66,10 +66,10 @@ func (a *adapter) TenantGroupSettingsCreate(id t.TenantID, settings *t.GroupSett
return nil, err
}
_, err = a.db.ExecContext(ctx, `INSERT INTO im_group_settings
(id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,created_by,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
(id,tenant_id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,created_by,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE name=VALUES(name),avatar_file_id=VALUES(avatar_file_id),description=VALUES(description),updated_at=VALUES(updated_at)`,
settingsID, settings.Topic, decodeUID(settings.OwnerUserID), settings.Name, settings.AvatarFileID, settings.Description, settings.GroupStatus,
settingsID, id, settings.Topic, decodeUID(settings.OwnerUserID), settings.Name, settings.AvatarFileID, settings.Description, settings.GroupStatus,
settings.JoinApproval, settings.MuteAll, settings.MemberUploadFiles, settings.MemberPrivateChat, settings.MemberInvite,
settings.MaxMemberCount, settings.MaxAdminCount, settings.MemberCount, settings.Version, decodeUID(settings.CreatedBy), settings.CreatedAt, settings.UpdatedAt)
if err != nil {
@@ -91,7 +91,7 @@ func (a *adapter) TenantGroupSettingsGet(id t.TenantID, topic string) (*t.GroupS
if cancel != nil {
defer cancel()
}
return a.groupSettingsByTopic(ctx, topic)
return a.groupSettingsByTopic(ctx, id, topic)
}
func (a *adapter) TenantGroupSettingsUpdate(id t.TenantID, topic string, patch t.GroupSettingsPatch) (*t.GroupSettings, error) {
@@ -154,8 +154,8 @@ func (a *adapter) TenantGroupSettingsUpdate(id t.TenantID, topic string, patch t
args = append(args, patch.MemberCountDelta)
}
set = append(set, "version=version+1")
args = append(args, topic)
where := "topic=?"
args = append(args, id, topic)
where := "tenant_id=? AND topic=?"
if patch.Version > 0 {
where += " AND version=?"
args = append(args, patch.Version)
@@ -187,7 +187,7 @@ func (a *adapter) TenantGroupSettingsUpdate(id t.TenantID, topic string, patch t
if err = tx.Commit(); err != nil {
return nil, err
}
return a.groupSettingsByTopic(ctx, topic)
return a.groupSettingsByTopic(ctx, id, topic)
}
func (a *adapter) TenantGroupDissolve(id t.TenantID, topic string, actor t.Uid, reason string, now time.Time) (*t.GroupSettings, error) {
@@ -213,7 +213,7 @@ func (a *adapter) TenantGroupDissolve(id t.TenantID, topic string, actor t.Uid,
defer tx.Rollback()
res, err := tx.ExecContext(ctx, `UPDATE im_group_settings
SET group_status='dissolved',dissolved_by=?,dissolved_at=?,member_count=0,version=version+1,updated_at=?
WHERE topic=? AND group_status<>'dissolved'`, decodeUID(actor), now, now, strings.TrimSpace(topic))
WHERE tenant_id=? AND topic=? AND group_status<>'dissolved'`, decodeUID(actor), now, now, id, strings.TrimSpace(topic))
if err != nil {
return nil, err
}
@@ -238,7 +238,7 @@ func (a *adapter) TenantGroupDissolve(id t.TenantID, topic string, actor t.Uid,
if err = tx.Commit(); err != nil {
return nil, err
}
return a.groupSettingsByTopic(ctx, topic)
return a.groupSettingsByTopic(ctx, id, topic)
}
func (a *adapter) TenantGroupMemberUpsert(id t.TenantID, member *t.GroupMember) (*t.GroupMember, error) {
@@ -1245,7 +1245,7 @@ func (a *adapter) TenantGroupCreate(id t.TenantID, topic *t.Topic, owner t.Uid,
return err
}
settings.MemberCount = 1
if err = insertGroupSettingsTx(tx, settings); err != nil {
if err = insertGroupSettingsTx(tx, id, settings); err != nil {
return err
}
ownerMember := &t.GroupMember{Topic: topic.Id, UserID: owner, Role: t.GroupMemberRoleOwner,
@@ -1458,8 +1458,8 @@ func (a *adapter) TenantGroupOwnerTransfer(id t.TenantID, topic string, previous
if _, err = tx.ExecContext(ctx, tenantShardSQL(id, "UPDATE im_topics SET owner=?,updatedat=? WHERE tenant_id=? AND name=?"), decodeUID(newOwner), now, id, topic); err != nil {
return err
}
args := []any{decodeUID(newOwner), now, topic}
query := "UPDATE im_group_settings SET owner_user_id=?,version=version+1,updated_at=? WHERE topic=?"
args := []any{decodeUID(newOwner), now, id, topic}
query := "UPDATE im_group_settings SET owner_user_id=?,version=version+1,updated_at=? WHERE tenant_id=? AND topic=?"
if version > 0 {
query += " AND version=?"
args = append(args, version)
@@ -1489,23 +1489,23 @@ func (a *adapter) TenantGroupCanPost(id t.TenantID, topic string, uid t.Uid, adm
}
var muteAll, memberMuted bool
err := a.db.QueryRowContext(ctx, `SELECT
COALESCE((SELECT mute_all FROM im_group_settings WHERE topic=? AND group_status=? LIMIT 1),0),
COALESCE((SELECT mute_all FROM im_group_settings WHERE tenant_id=? AND topic=? AND group_status=? LIMIT 1),0),
EXISTS(SELECT 1 FROM im_group_mutes WHERE topic=? AND user_id=? AND status=? AND (expires_at IS NULL OR expires_at>?))`,
topic, t.GroupStatusActive, topic, decodeUID(uid), t.GroupMuteStatusActive, now).Scan(&muteAll, &memberMuted)
id, topic, t.GroupStatusActive, topic, decodeUID(uid), t.GroupMuteStatusActive, now).Scan(&muteAll, &memberMuted)
if err != nil {
return false, err
}
return !muteAll && !memberMuted, nil
}
func insertGroupSettingsTx(tx *sqlx.Tx, settings *t.GroupSettings) error {
func insertGroupSettingsTx(tx *sqlx.Tx, tenantID t.TenantID, settings *t.GroupSettings) error {
settingsID, err := nextSnowflakeID()
if err != nil {
return err
}
_, err = tx.Exec(`INSERT INTO im_group_settings
(id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,created_by,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, settingsID, settings.Topic, decodeUID(settings.OwnerUserID), settings.Name,
(id,tenant_id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,created_by,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, settingsID, tenantID, settings.Topic, decodeUID(settings.OwnerUserID), settings.Name,
settings.AvatarFileID, settings.Description, settings.GroupStatus, settings.JoinApproval, settings.MuteAll,
settings.MemberUploadFiles, settings.MemberPrivateChat, settings.MemberInvite, settings.MaxMemberCount,
settings.MaxAdminCount, settings.MemberCount, settings.Version, decodeUID(settings.CreatedBy), settings.CreatedAt, settings.UpdatedAt)
@@ -1569,9 +1569,9 @@ func (a *adapter) routeGroupTenant(id t.TenantID) (*adapter, error) {
return a, validTenant(id)
}
func (a *adapter) groupSettingsByTopic(ctx context.Context, topic string) (*t.GroupSettings, error) {
func (a *adapter) groupSettingsByTopic(ctx context.Context, tenantID t.TenantID, topic string) (*t.GroupSettings, error) {
var item t.GroupSettings
err := a.db.QueryRowxContext(ctx, "SELECT "+groupSettingsSelectColumns+" FROM im_group_settings WHERE topic=?", topic).StructScan(&item)
err := a.db.QueryRowxContext(ctx, "SELECT "+groupSettingsSelectColumns+" FROM im_group_settings WHERE tenant_id=? AND topic=?", tenantID, topic).StructScan(&item)
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}