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

feat: shard tenant topic and membership tables

Browse Source
This commit is contained in:
richjohnrambo
2026-09-23 02:14:13 +08:00
parent 432e7ac51a
commit 22adc13acf
3 changed files with 110 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/logic/sys/im_group.go
+39 -34
View File
@@ -68,7 +68,7 @@ func (s *sSysImGroup) List(ctx context.Context, in *sysin.ImGroupListInp) (list
} }
func (s *sSysImGroup) groupModel(ctx context.Context, db gdb.DB, tenantId int64) *gdb.Model { func (s *sSysImGroup) groupModel(ctx context.Context, db gdb.DB, tenantId int64) *gdb.Model {
return db.Model("im_topics t").Safe().Ctx(ctx). return db.Model(tenantTablePrefix(tenantId)+"topics t").Safe().Ctx(ctx).
LeftJoin("im_group_settings gs", "gs.tenant_id = t.tenant_id AND gs.topic = t.name"). LeftJoin("im_group_settings gs", "gs.tenant_id = t.tenant_id AND gs.topic = t.name").
LeftJoin("im_users owner", "owner.tenant_id = t.tenant_id AND owner.id = COALESCE(gs.owner_user_id, t.owner)"). LeftJoin("im_users owner", "owner.tenant_id = t.tenant_id AND owner.id = COALESCE(gs.owner_user_id, t.owner)").
LeftJoin("im_user_profiles owner_profile", "owner_profile.tenant_id = t.tenant_id AND owner_profile.user_id = COALESCE(gs.owner_user_id, t.owner)"). LeftJoin("im_user_profiles owner_profile", "owner_profile.tenant_id = t.tenant_id AND owner_profile.user_id = COALESCE(gs.owner_user_id, t.owner)").
@@ -140,7 +140,7 @@ func (s *sSysImGroup) View(ctx context.Context, in *sysin.ImGroupViewInp) (res *
return return
} }
err = db.Model("im_topics t").Safe().Ctx(ctx). err = db.Model(tenantTablePrefix(tenant.Id)+"topics t").Safe().Ctx(ctx).
LeftJoin("im_group_settings gs", "gs.tenant_id = t.tenant_id AND gs.topic = t.name"). LeftJoin("im_group_settings gs", "gs.tenant_id = t.tenant_id AND gs.topic = t.name").
LeftJoin("im_users owner", "owner.tenant_id = t.tenant_id AND owner.id = COALESCE(gs.owner_user_id, t.owner)"). LeftJoin("im_users owner", "owner.tenant_id = t.tenant_id AND owner.id = COALESCE(gs.owner_user_id, t.owner)").
LeftJoin("im_user_profiles owner_profile", "owner_profile.tenant_id = t.tenant_id AND owner_profile.user_id = COALESCE(gs.owner_user_id, t.owner)"). LeftJoin("im_user_profiles owner_profile", "owner_profile.tenant_id = t.tenant_id AND owner_profile.user_id = COALESCE(gs.owner_user_id, t.owner)").
@@ -210,15 +210,15 @@ func (s *sSysImGroup) MemberList(ctx context.Context, in *sysin.ImGroupMemberLis
} }
func (s *sSysImGroup) hasGroupMemberTruth(ctx context.Context, db gdb.DB, tenantId int64, topic string) bool { func (s *sSysImGroup) hasGroupMemberTruth(ctx context.Context, db gdb.DB, tenantId int64, topic string) bool {
count, err := db.Model("im_group_members").Safe().Ctx(ctx).Where("tenant_id", tenantId).Where("topic", topic).Count() count, err := db.Model(tenantTablePrefix(tenantId)+"group_members").Safe().Ctx(ctx).Where("tenant_id", tenantId).Where("topic", topic).Count()
return err == nil && count > 0 return err == nil && count > 0
} }
func (s *sSysImGroup) memberListFromBusiness(ctx context.Context, db gdb.DB, tenantId int64, in *sysin.ImGroupMemberListInp) (list []*sysin.ImGroupMemberListModel, totalCount int, err error) { func (s *sSysImGroup) memberListFromBusiness(ctx context.Context, db gdb.DB, tenantId int64, in *sysin.ImGroupMemberListInp) (list []*sysin.ImGroupMemberListModel, totalCount int, err error) {
mod := db.Model("im_group_members gm").Safe().Ctx(ctx). mod := db.Model(tenantTablePrefix(tenantId)+"group_members gm").Safe().Ctx(ctx).
LeftJoin("im_users u", "u.tenant_id = gm.tenant_id AND u.id = gm.user_id"). LeftJoin("im_users u", "u.tenant_id = gm.tenant_id AND u.id = gm.user_id").
LeftJoin("im_user_profiles up", "up.tenant_id = gm.tenant_id AND up.user_id = gm.user_id"). LeftJoin("im_user_profiles up", "up.tenant_id = gm.tenant_id AND up.user_id = gm.user_id").
LeftJoin("im_subscriptions s", "s.tenant_id = gm.tenant_id AND s.topic = gm.topic AND s.userid = gm.user_id"). LeftJoin(tenantTablePrefix(tenantId)+"subscriptions s", "s.tenant_id = gm.tenant_id AND s.topic = gm.topic AND s.userid = gm.user_id").
Fields( Fields(
"gm.user_id", "gm.user_id",
"up.nickname", "up.nickname",
@@ -264,7 +264,7 @@ func (s *sSysImGroup) memberListFromBusiness(ctx context.Context, db gdb.DB, ten
} }
func (s *sSysImGroup) memberListFromProjection(ctx context.Context, db gdb.DB, tenantId int64, in *sysin.ImGroupMemberListInp) (list []*sysin.ImGroupMemberListModel, totalCount int, err error) { func (s *sSysImGroup) memberListFromProjection(ctx context.Context, db gdb.DB, tenantId int64, in *sysin.ImGroupMemberListInp) (list []*sysin.ImGroupMemberListModel, totalCount int, err error) {
mod := db.Model("im_subscriptions s").Safe().Ctx(ctx). mod := db.Model(tenantTablePrefix(tenantId)+"subscriptions s").Safe().Ctx(ctx).
LeftJoin("im_users u", "u.tenant_id = s.tenant_id AND u.id = s.userid"). LeftJoin("im_users u", "u.tenant_id = s.tenant_id AND u.id = s.userid").
LeftJoin("im_user_profiles up", "up.tenant_id = s.tenant_id AND up.user_id = s.userid"). LeftJoin("im_user_profiles up", "up.tenant_id = s.tenant_id AND up.user_id = s.userid").
Fields( Fields(
@@ -471,7 +471,7 @@ func (s *sSysImGroup) JoinRequestReject(ctx context.Context, in *sysin.ImGroupJo
} }
func (s *sSysImGroup) resolveJoinRequest(ctx context.Context, in *sysin.ImGroupJoinRequestResolveInp, status string) (err error) { func (s *sSysImGroup) resolveJoinRequest(ctx context.Context, in *sysin.ImGroupJoinRequestResolveInp, status string) (err error) {
db, _, err := resolveImTenantDB(ctx, in.TenantId) db, tenant, err := resolveImTenantDB(ctx, in.TenantId)
if err != nil { if err != nil {
return err return err
} }
@@ -497,10 +497,10 @@ func (s *sSysImGroup) resolveJoinRequest(ctx context.Context, in *sysin.ImGroupJ
} }
applicantUserId := before["applicant_user_id"].Int64() applicantUserId := before["applicant_user_id"].Int64()
if status == consts.ImGroupRequestStatusAccepted { if status == consts.ImGroupRequestStatusAccepted {
if txErr = s.upsertGroupSubscription(ctx, tx, in.Topic, applicantUserId, imGroupModeMember, now); txErr != nil { if txErr = s.upsertGroupSubscription(ctx, tx, tenant.Id, in.Topic, applicantUserId, imGroupModeMember, now); txErr != nil {
return txErr return txErr
} }
if txErr = s.upsertGroupMember(ctx, tx, in.Topic, applicantUserId, consts.ImGroupRoleMember, "approval", actor, now); txErr != nil { if txErr = s.upsertGroupMember(ctx, tx, tenant.Id, in.Topic, applicantUserId, consts.ImGroupRoleMember, "approval", actor, now); txErr != nil {
return txErr return txErr
} }
} }
@@ -514,7 +514,7 @@ func (s *sSysImGroup) resolveJoinRequest(ctx context.Context, in *sysin.ImGroupJ
if affected, _ := result.RowsAffected(); affected == 0 { if affected, _ := result.RowsAffected(); affected == 0 {
return gerror.New("入群申请已处理") return gerror.New("入群申请已处理")
} }
if txErr = s.recountGroupMembers(ctx, tx, in.Topic); txErr != nil { if txErr = s.recountGroupMembers(ctx, tx, tenant.Id, in.Topic); txErr != nil {
return txErr return txErr
} }
after, txErr := s.groupJoinRequestSnapshot(ctx, tx, in.Topic, in.RequestId) after, txErr := s.groupJoinRequestSnapshot(ctx, tx, in.Topic, in.RequestId)
@@ -644,8 +644,8 @@ func (s *sSysImGroup) groupJoinRequestSnapshot(ctx context.Context, tx gdb.TX, t
One() One()
} }
func (s *sSysImGroup) groupMemberSnapshot(ctx context.Context, tx gdb.TX, topic string, userId int64) (res gdb.Record, err error) { func (s *sSysImGroup) groupMemberSnapshot(ctx context.Context, tx gdb.TX, tenantId int64, topic string, userId int64) (res gdb.Record, err error) {
return tx.Model("im_group_members").Safe().Ctx(ctx). return tx.Model(tenantTablePrefix(tenantId)+"group_members").Safe().Ctx(ctx).
Where("topic", topic). Where("topic", topic).
Where("user_id", userId). Where("user_id", userId).
One() One()
@@ -683,8 +683,8 @@ func (s *sSysImGroup) groupMuteSnapshot(ctx context.Context, tx gdb.TX, topic st
One() One()
} }
func (s *sSysImGroup) ensureActiveGroupMember(ctx context.Context, tx gdb.TX, topic string, userId int64) error { func (s *sSysImGroup) ensureActiveGroupMember(ctx context.Context, tx gdb.TX, tenantId int64, topic string, userId int64) error {
count, err := tx.Model("im_group_members").Safe().Ctx(ctx). count, err := tx.Model(tenantTablePrefix(tenantId)+"group_members").Safe().Ctx(ctx).
Where("topic", topic). Where("topic", topic).
Where("user_id", userId). Where("user_id", userId).
Where("member_status", consts.ImGroupMemberStatusActive). Where("member_status", consts.ImGroupMemberStatusActive).
@@ -698,8 +698,8 @@ func (s *sSysImGroup) ensureActiveGroupMember(ctx context.Context, tx gdb.TX, to
return nil return nil
} }
func (s *sSysImGroup) ensureAdminLimit(ctx context.Context, tx gdb.TX, topic string, userId int64) error { func (s *sSysImGroup) ensureAdminLimit(ctx context.Context, tx gdb.TX, tenantId int64, topic string, userId int64) error {
currentRole, err := tx.Model("im_group_members").Safe().Ctx(ctx). currentRole, err := tx.Model(tenantTablePrefix(tenantId)+"group_members").Safe().Ctx(ctx).
Where("topic", topic). Where("topic", topic).
Where("user_id", userId). Where("user_id", userId).
Where("member_status", consts.ImGroupMemberStatusActive). Where("member_status", consts.ImGroupMemberStatusActive).
@@ -717,7 +717,7 @@ func (s *sSysImGroup) ensureAdminLimit(ctx context.Context, tx gdb.TX, topic str
if maxAdminCount.Int() <= 0 { if maxAdminCount.Int() <= 0 {
return nil return nil
} }
adminCount, err := tx.Model("im_group_members").Safe().Ctx(ctx). adminCount, err := tx.Model(tenantTablePrefix(tenantId)+"group_members").Safe().Ctx(ctx).
Where("topic", topic). Where("topic", topic).
Where("role", consts.ImGroupRoleAdmin). Where("role", consts.ImGroupRoleAdmin).
Where("member_status", consts.ImGroupMemberStatusActive). Where("member_status", consts.ImGroupMemberStatusActive).
@@ -731,33 +731,38 @@ func (s *sSysImGroup) ensureAdminLimit(ctx context.Context, tx gdb.TX, topic str
return nil return nil
} }
func (s *sSysImGroup) upsertGroupSubscription(ctx context.Context, tx gdb.TX, topic string, userId int64, modeGiven string, now *gtime.Time) error { func (s *sSysImGroup) upsertGroupSubscription(ctx context.Context, tx gdb.TX, tenantId int64, topic string, userId int64, modeGiven string, now *gtime.Time) error {
_, err := tx.Exec(`INSERT INTO im_subscriptions(createdat,updatedat,deletedat,userid,topic,modewant,modegiven,private) table := tenantTablePrefix(tenantId) + "subscriptions"
VALUES(?,?,NULL,?,?,?,?,CAST('{}' AS JSON)) _, err := tx.Exec(fmt.Sprintf(`INSERT INTO %s(tenant_id,createdat,updatedat,deletedat,userid,topic,modewant,modegiven,private)
VALUES(?,?,?,NULL,?,?,?,?,CAST('{}' AS JSON))
ON DUPLICATE KEY UPDATE createdat=VALUES(createdat),updatedat=VALUES(updatedat),deletedat=NULL, ON DUPLICATE KEY UPDATE createdat=VALUES(createdat),updatedat=VALUES(updatedat),deletedat=NULL,
modewant=VALUES(modewant),modegiven=VALUES(modegiven),delid=0,recvseqid=0,readseqid=0`, modewant=VALUES(modewant),modegiven=VALUES(modegiven),delid=0,recvseqid=0,readseqid=0`, table),
now, now, userId, topic, imGroupModeMember, modeGiven) tenantId, now, now, userId, topic, imGroupModeMember, modeGiven)
return err return err
} }
func (s *sSysImGroup) upsertGroupMember(ctx context.Context, tx gdb.TX, topic string, userId int64, role string, joinSource string, joinedBy int64, now *gtime.Time) error { func (s *sSysImGroup) upsertGroupMember(ctx context.Context, tx gdb.TX, tenantId int64, topic string, userId int64, role string, joinSource string, joinedBy int64, now *gtime.Time) error {
_, err := tx.Exec(`INSERT INTO im_group_members(topic,user_id,role,member_status,join_source,joined_by,joined_at,version,created_at,updated_at) table := tenantTablePrefix(tenantId) + "group_members"
VALUES(?,?,?,?,?,?,?,?,?,?) _, err := tx.Exec(fmt.Sprintf(`INSERT INTO %s(tenant_id,topic,user_id,role,member_status,join_source,joined_by,joined_at,version,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE role=VALUES(role),member_status=VALUES(member_status),join_source=VALUES(join_source), ON DUPLICATE KEY UPDATE role=VALUES(role),member_status=VALUES(member_status),join_source=VALUES(join_source),
joined_by=VALUES(joined_by),joined_at=VALUES(joined_at),left_at=NULL,removed_by=NULL,version=version+1,updated_at=VALUES(updated_at)`, joined_by=VALUES(joined_by),joined_at=VALUES(joined_at),left_at=NULL,removed_by=NULL,version=version+1,updated_at=VALUES(updated_at)`, table),
topic, userId, role, consts.ImGroupMemberStatusActive, joinSource, joinedBy, now, 1, now, now) tenantId, topic, userId, role, consts.ImGroupMemberStatusActive, joinSource, joinedBy, now, 1, now, now)
return err return err
} }
func (s *sSysImGroup) recountGroupMembers(ctx context.Context, tx gdb.TX, topic string) error { func (s *sSysImGroup) recountGroupMembers(ctx context.Context, tx gdb.TX, tenantId int64, topic string) error {
if _, err := tx.Exec(`UPDATE im_group_settings membersTable := tenantTablePrefix(tenantId) + "group_members"
SET member_count=(SELECT COUNT(*) FROM im_group_members WHERE topic=? AND member_status=?),updated_at=CURRENT_TIMESTAMP(3) subscriptionsTable := tenantTablePrefix(tenantId) + "subscriptions"
WHERE topic=?`, topic, consts.ImGroupMemberStatusActive, topic); err != nil { topicsTable := tenantTablePrefix(tenantId) + "topics"
if _, err := tx.Exec(fmt.Sprintf(`UPDATE im_group_settings
SET member_count=(SELECT COUNT(*) FROM %s WHERE tenant_id=? AND topic=? AND member_status=?),updated_at=CURRENT_TIMESTAMP(3)
WHERE tenant_id=? AND topic=?`, membersTable), tenantId, topic, consts.ImGroupMemberStatusActive, tenantId, topic); err != nil {
return err return err
} }
_, err := tx.Exec(`UPDATE im_topics _, err := tx.Exec(fmt.Sprintf(`UPDATE %s
SET subcnt=(SELECT COUNT(*) FROM im_subscriptions WHERE topic=? AND deletedat IS NULL),updatedat=CURRENT_TIMESTAMP(3) SET subcnt=(SELECT COUNT(*) FROM %s WHERE tenant_id=? AND topic=? AND deletedat IS NULL),updatedat=CURRENT_TIMESTAMP(3)
WHERE name=?`, topic, topic) WHERE tenant_id=? AND name=?`, topicsTable, subscriptionsTable), tenantId, topic, tenantId, topic)
return err return err
} }
server/internal/logic/sys/im_tenant_tables.go
+4
View File
@@ -59,6 +59,10 @@ func provisionTenantMessageTables(ctx context.Context, tenantID int64) error {
baseTables := [][2]string{ baseTables := [][2]string{
{prefix + "friendships", "im_friendships"}, {prefix + "friendships", "im_friendships"},
{prefix + "messages", "im_messages"}, {prefix + "messages", "im_messages"},
{prefix + "topics", "im_topics"},
{prefix + "group_members", "im_group_members"},
{prefix + "subscriptions", "im_subscriptions"},
{prefix + "system_notifications", "im_system_notifications"},
} }
for _, pair := range baseTables { for _, pair := range baseTables {
if _, err = db.Exec(ctx, fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` LIKE `%s`", pair[0], pair[1])); err != nil { if _, err = db.Exec(ctx, fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` LIKE `%s`", pair[0], pair[1])); err != nil {
server/storage/data/generate/im_additional_tenant_shards.sql
+67
View File
@@ -0,0 +1,67 @@
-- Existing-merchant migration for topics, group members, subscriptions and system notifications.
-- Run against meeting_im. The script is idempotent and keeps the shared source tables intact.
CREATE TABLE IF NOT EXISTS im_topics_bak_20260923_shard LIKE im_topics;
INSERT IGNORE INTO im_topics_bak_20260923_shard SELECT * FROM im_topics;
CREATE TABLE IF NOT EXISTS im_group_members_bak_20260923_shard LIKE im_group_members;
INSERT IGNORE INTO im_group_members_bak_20260923_shard SELECT * FROM im_group_members;
CREATE TABLE IF NOT EXISTS im_subscriptions_bak_20260923_shard LIKE im_subscriptions;
INSERT IGNORE INTO im_subscriptions_bak_20260923_shard SELECT * FROM im_subscriptions;
CREATE TABLE IF NOT EXISTS im_system_notifications_bak_20260923_shard LIKE im_system_notifications;
INSERT IGNORE INTO im_system_notifications_bak_20260923_shard SELECT * FROM im_system_notifications;
DROP PROCEDURE IF EXISTS migrate_additional_tenant_shards;
DELIMITER $$
CREATE PROCEDURE migrate_additional_tenant_shards()
BEGIN
DECLARE done INT DEFAULT 0;
DECLARE current_tenant_id BIGINT;
DECLARE tenant_cursor CURSOR FOR SELECT id FROM hotgo.im_tenant ORDER BY id;
DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = 1;
OPEN tenant_cursor;
tenant_loop: LOOP
FETCH tenant_cursor INTO current_tenant_id;
IF done = 1 THEN
LEAVE tenant_loop;
END IF;
SET @tenant_prefix = CONCAT('im_', LPAD(current_tenant_id, 3, '0'), '_');
SET @sql = CONCAT('CREATE TABLE IF NOT EXISTS `', @tenant_prefix, 'topics` LIKE `im_topics`');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('ALTER TABLE `', @tenant_prefix, 'topics` MODIFY tenant_id BIGINT NOT NULL DEFAULT ', current_tenant_id, ' COMMENT ''商户ID''');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('REPLACE INTO `', @tenant_prefix, 'topics` SELECT * FROM `im_topics` WHERE tenant_id=', current_tenant_id);
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('CREATE TABLE IF NOT EXISTS `', @tenant_prefix, 'group_members` LIKE `im_group_members`');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('ALTER TABLE `', @tenant_prefix, 'group_members` MODIFY tenant_id BIGINT NOT NULL DEFAULT ', current_tenant_id, ' COMMENT ''商户ID''');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('REPLACE INTO `', @tenant_prefix, 'group_members` SELECT * FROM `im_group_members` WHERE tenant_id=', current_tenant_id);
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('CREATE TABLE IF NOT EXISTS `', @tenant_prefix, 'subscriptions` LIKE `im_subscriptions`');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('ALTER TABLE `', @tenant_prefix, 'subscriptions` MODIFY tenant_id BIGINT NOT NULL DEFAULT ', current_tenant_id, ' COMMENT ''商户ID''');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('REPLACE INTO `', @tenant_prefix, 'subscriptions` SELECT * FROM `im_subscriptions` WHERE tenant_id=', current_tenant_id);
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('CREATE TABLE IF NOT EXISTS `', @tenant_prefix, 'system_notifications` LIKE `im_system_notifications`');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('ALTER TABLE `', @tenant_prefix, 'system_notifications` MODIFY tenant_id BIGINT NOT NULL DEFAULT ', current_tenant_id, ' COMMENT ''商户ID''');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @sql = CONCAT('REPLACE INTO `', @tenant_prefix, 'system_notifications` SELECT * FROM `im_system_notifications` WHERE tenant_id=', current_tenant_id);
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
END LOOP;
CLOSE tenant_cursor;
END$$
DELIMITER ;
CALL migrate_additional_tenant_shards();
DROP PROCEDURE migrate_additional_tenant_shards;