From 851eca97ed875b39f689ee07b3eb3f30e61e9e86 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Tue, 22 Sep 2026 15:19:56 +0800 Subject: [PATCH] feat: support shared tenant database sharding --- server/internal/library/imtenant/db.go | 41 ++++-- server/internal/logic/sys/im_group.go | 47 ++++--- server/internal/logic/sys/im_tenant.go | 69 ++++++--- server/internal/logic/sys/im_tenant_tables.go | 131 ++++++++++++++++++ server/internal/logic/sys/im_user.go | 52 +++---- .../internal/model/input/sysin/im_tenant.go | 1 + 6 files changed, 265 insertions(+), 76 deletions(-) create mode 100644 server/internal/logic/sys/im_tenant_tables.go diff --git a/server/internal/library/imtenant/db.go b/server/internal/library/imtenant/db.go index 04a8b7a..e774f14 100644 --- a/server/internal/library/imtenant/db.go +++ b/server/internal/library/imtenant/db.go @@ -23,7 +23,7 @@ const registryDbGroup = "imRegistry" var registryDbGroupOnce sync.Once -// DB resolves a tenant registry record and returns the corresponding business DB. +// DB resolves a tenant registry record and returns the shared IM business DB. func DB(ctx context.Context, tenantId int64) (db gdb.DB, tenant *entity.ImTenant, err error) { if tenantId <= 0 { return nil, nil, gerror.New("租户ID不能为空") @@ -36,27 +36,48 @@ func DB(ctx context.Context, tenantId int64) (db gdb.DB, tenant *entity.ImTenant return nil, nil, gerror.New("租户不存在") } - link, err := tenantLink(ctx, tenant) + db, err = SharedDB(ctx) if err != nil { return nil, nil, err } + return db, tenant, nil +} - group := fmt.Sprintf("im_tenant_%d_%x", tenant.Id, hash(link)) +// SharedDB returns the common IM database configured by the default tenant. +// Tenant isolation inside this database is implemented by tenant_id predicates +// and the three tenant-partitioned high-volume table families. +func SharedDB(ctx context.Context) (gdb.DB, error) { + var sharedTenant *entity.ImTenant + if err := RegistryDB().Model(dao.ImTenant.Table()).Safe().Ctx(ctx). + Where(dao.ImTenant.Columns().Code, "default").Scan(&sharedTenant); err != nil { + return nil, gerror.Wrap(err, "获取共享业务库配置失败") + } + if sharedTenant == nil || sharedTenant.Id <= 0 { + return nil, gerror.New("默认商户不存在，无法确定共享业务库") + } + link, err := tenantLink(ctx, sharedTenant) + if err != nil { + return nil, err + } + return dbForLink(link, sharedTenant.Id) +} + +func dbForLink(link string, ownerID int64) (gdb.DB, error) { + group := fmt.Sprintf("im_shared_%d_%x", ownerID, hash(link)) if cached, ok := tenantDbGroups.Load(group); ok && cached == link { - db, err = gdb.Instance(group) - return db, tenant, err + return gdb.Instance(group) } - if err = gdb.SetConfigGroup(group, gdb.ConfigGroup{{Link: link}}); err != nil { - return nil, nil, gerror.Wrap(err, "初始化租户业务库连接失败") + if err := gdb.SetConfigGroup(group, gdb.ConfigGroup{{Link: link}}); err != nil { + return nil, gerror.Wrap(err, "初始化共享业务库连接失败") } tenantDbGroups.Store(group, link) - db, err = gdb.Instance(group) + db, err := gdb.Instance(group) if err != nil { - return nil, nil, gerror.Wrap(err, "获取租户业务库连接失败") + return nil, gerror.Wrap(err, "获取共享业务库连接失败") } - return db, tenant, nil + return db, nil } // RegistryDB returns the dedicated tenant registry connection, or the default diff --git a/server/internal/logic/sys/im_group.go b/server/internal/logic/sys/im_group.go index 8925fc8..9b2fd48 100644 --- a/server/internal/logic/sys/im_group.go +++ b/server/internal/logic/sys/im_group.go @@ -52,7 +52,7 @@ func (s *sSysImGroup) List(ctx context.Context, in *sysin.ImGroupListInp) (list return } - mod := s.groupModel(ctx, db) + mod := s.groupModel(ctx, db, tenant.Id) mod = s.applyGroupWhere(mod, in) mod = mod.Page(in.Page, in.PerPage).OrderDesc("t.touchedat").OrderDesc("t.id") @@ -67,11 +67,11 @@ func (s *sSysImGroup) List(ctx context.Context, in *sysin.ImGroupListInp) (list return } -func (s *sSysImGroup) groupModel(ctx context.Context, db gdb.DB) *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). - LeftJoin("im_group_settings gs", "gs.topic = t.name"). - LeftJoin("im_users owner", "owner.id = COALESCE(gs.owner_user_id, t.owner)"). - LeftJoin("im_user_profiles owner_profile", "owner_profile.user_id = COALESCE(gs.owner_user_id, t.owner)"). + 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_user_profiles owner_profile", "owner_profile.tenant_id = t.tenant_id AND owner_profile.user_id = COALESCE(gs.owner_user_id, t.owner)"). Fields( "t.name topic", "t.state", @@ -103,6 +103,7 @@ func (s *sSysImGroup) groupModel(ctx context.Context, db gdb.DB) *gdb.Model { "COALESCE(gs.version, 0) settings_version", "gs.updated_at settings_updated_at", ). + Where("t.tenant_id", tenantId). WhereLike("t.name", "grp%") } @@ -140,9 +141,9 @@ func (s *sSysImGroup) View(ctx context.Context, in *sysin.ImGroupViewInp) (res * } err = db.Model("im_topics t").Safe().Ctx(ctx). - LeftJoin("im_group_settings gs", "gs.topic = t.name"). - LeftJoin("im_users owner", "owner.id = COALESCE(gs.owner_user_id, t.owner)"). - LeftJoin("im_user_profiles owner_profile", "owner_profile.user_id = COALESCE(gs.owner_user_id, t.owner)"). + 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_user_profiles owner_profile", "owner_profile.tenant_id = t.tenant_id AND owner_profile.user_id = COALESCE(gs.owner_user_id, t.owner)"). Fields( "t.name topic", "t.state", @@ -176,6 +177,7 @@ func (s *sSysImGroup) View(ctx context.Context, in *sysin.ImGroupViewInp) (res * "COALESCE(gs.version, 0) settings_version", "gs.updated_at settings_updated_at", ). + Where("t.tenant_id", tenant.Id). Where("t.name", in.Topic). WhereLike("t.name", "grp%"). Scan(&res) @@ -196,27 +198,27 @@ func (s *sSysImGroup) MemberList(ctx context.Context, in *sysin.ImGroupMemberLis return } - db, _, err := resolveImTenantDB(ctx, in.TenantId) + db, tenant, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return } - if s.hasGroupMemberTruth(ctx, db, in.Topic) { - return s.memberListFromBusiness(ctx, db, in) + if s.hasGroupMemberTruth(ctx, db, tenant.Id, in.Topic) { + return s.memberListFromBusiness(ctx, db, tenant.Id, in) } - return s.memberListFromProjection(ctx, db, in) + return s.memberListFromProjection(ctx, db, tenant.Id, in) } -func (s *sSysImGroup) hasGroupMemberTruth(ctx context.Context, db gdb.DB, topic string) bool { - count, err := db.Model("im_group_members").Safe().Ctx(ctx).Where("topic", topic).Count() +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() return err == nil && count > 0 } -func (s *sSysImGroup) memberListFromBusiness(ctx context.Context, db gdb.DB, 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). - LeftJoin("im_users u", "u.id = gm.user_id"). - LeftJoin("im_user_profiles up", "up.user_id = gm.user_id"). - LeftJoin("im_subscriptions s", "s.topic = gm.topic AND s.userid = 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_subscriptions s", "s.tenant_id = gm.tenant_id AND s.topic = gm.topic AND s.userid = gm.user_id"). Fields( "gm.user_id", "up.nickname", @@ -242,6 +244,7 @@ func (s *sSysImGroup) memberListFromBusiness(ctx context.Context, db gdb.DB, in "gm.removed_by", "gm.version", ). + Where("gm.tenant_id", tenantId). Where("gm.topic", in.Topic) if in.UserId > 0 { mod = mod.Where("gm.user_id", in.UserId) @@ -260,10 +263,10 @@ func (s *sSysImGroup) memberListFromBusiness(ctx context.Context, db gdb.DB, in return } -func (s *sSysImGroup) memberListFromProjection(ctx context.Context, db gdb.DB, 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). - LeftJoin("im_users u", "u.id = s.userid"). - LeftJoin("im_user_profiles up", "up.user_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"). Fields( "s.userid user_id", "up.nickname", @@ -289,6 +292,7 @@ func (s *sSysImGroup) memberListFromProjection(ctx context.Context, db gdb.DB, i "0 removed_by", "0 version", ). + Where("s.tenant_id", tenantId). Where("s.topic", in.Topic) if in.UserId > 0 { @@ -335,6 +339,7 @@ func (s *sSysImGroup) Governance(ctx context.Context, in *sysin.ImGroupTopicInp) "created_at", "updated_at", ). + Where("tenant_id", tenant.Id). Where("topic", in.Topic). Scan(&res) if err != nil { diff --git a/server/internal/logic/sys/im_tenant.go b/server/internal/logic/sys/im_tenant.go index 23baa75..442f127 100644 --- a/server/internal/logic/sys/im_tenant.go +++ b/server/internal/logic/sys/im_tenant.go @@ -222,13 +222,23 @@ func (s *sSysImTenant) saveTenant(ctx context.Context, in *sysin.ImTenantEditInp super := isSuperTenantAdmin(ctx) isCreate := in.Id == 0 tenantID := in.Id - - // 业务库名与独立连接串仅超管可维护，非超管提交一律强制忽略 - if !super { + requestedState := in.State + if isCreate { + if requestedState != consts.ImTenantStateActive && requestedState != consts.ImTenantStateSuspended { + return gerror.New("新商户创建完成后的状态只能是正常或已停用") + } + if err = ensureTenantIDSequence(ctx); err != nil { + return err + } + if in.DbName, in.DbDsn, err = sharedDatabaseConfig(ctx); err != nil { + return err + } + in.State = consts.ImTenantStateProvisioning + } else if !super { + // 业务库名与独立连接串仅超管可维护，非超管提交一律强制忽略。 in.DbName = "" in.DbDsn = "" } - err = imtenant.RegistryDB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) (err error) { if err = s.checkCodeUnique(ctx, in.Id, in.Code); err != nil { return @@ -255,16 +265,19 @@ func (s *sSysImTenant) saveTenant(ctx context.Context, in *sysin.ImTenantEditInp return s.saveConfig(ctx, in, userId) } + in.Id, err = nextTenantID(tx) + if err != nil { + return err + } in.CreatedBy = userId in.UpdatedBy = userId - id, err := s.Model(ctx, &handler.Option{FilterAuth: false}). + _, err = tx.Model(dao.ImTenant.Table()).Safe().Ctx(ctx). Fields(sysin.ImTenantInsertFields{}). - Data(in).InsertAndGetId() + Data(in).Insert() if err != nil { return gerror.Wrap(err, "新增IM租户失败，请稍后重试！") } - tenantID = id - in.Id = id + tenantID = in.Id return s.saveConfig(ctx, in, userId) }) if err != nil { @@ -277,13 +290,22 @@ func (s *sSysImTenant) saveTenant(ctx context.Context, in *sysin.ImTenantEditInp s.rollbackTenantCreation(ctx, tenantID) return err } + if err = provisionTenantMessageTables(ctx, tenantID); err != nil { + return gerror.Wrap(err, "商户已登记为初始化中，但业务分表创建失败") + } + if _, err = s.Model(ctx, &handler.Option{FilterAuth: false}).WherePri(tenantID).Data(g.Map{ + dao.ImTenant.Columns().State: requestedState, + dao.ImTenant.Columns().UpdatedBy: contexts.GetUserId(ctx), + }).Update(); err != nil { + return gerror.Wrap(err, "商户分表已创建，但更新商户状态失败") + } + in.State = requestedState } else if in.AdminPassword != "" { - // 编辑商户时填写了新密码则重置管理员密码，留空不修改 + // 编辑商户时填写了新密码则重置管理员密码，留空不修改。 if err = s.resetTenantAdminPassword(ctx, tenantID, in.AdminPassword); err != nil { return err } } - _ = publishImPolicyEvent(ctx, tenantID, in.Code, "tenant.features.updated", "moderation", 0) return nil } @@ -550,6 +572,11 @@ func (s *sSysImTenant) Status(ctx context.Context, in *sysin.ImTenantStatusInp) if err = in.Filter(ctx); err != nil { return } + if in.State == consts.ImTenantStateActive { + if err = provisionTenantMessageTables(ctx, in.Id); err != nil { + return gerror.Wrap(err, "启用商户前创建业务分表失败") + } + } if _, err = s.Model(ctx).WherePri(in.Id).Data(g.Map{ dao.ImTenant.Columns().State: in.State, dao.ImTenant.Columns().UpdatedBy: contexts.GetUserId(ctx), @@ -601,18 +628,19 @@ func (s *sSysImTenant) Usage(ctx context.Context, in *sysin.ImTenantUsageInp) (r res.DbReachable = true res.SchemaVersion = s.queryScalarString(ctx, db, "kvmeta.version", "SELECT value FROM kvmeta WHERE key_='version' LIMIT 1", &res.Warnings) - res.UserCount = s.queryScalarInt64(ctx, db, "im_users.total", "SELECT COUNT(*) FROM im_users", &res.Warnings) - res.ActiveUserCount = s.queryScalarInt64(ctx, db, "im_users.active", "SELECT COUNT(*) FROM im_users WHERE state=0", &res.Warnings) - res.GroupCount = s.queryScalarInt64(ctx, db, "im_group_settings", "SELECT COUNT(*) FROM im_group_settings WHERE group_status<>'dissolved'", &res.Warnings) - res.DeviceCount = s.queryScalarInt64(ctx, db, "im_devices", "SELECT COUNT(*) FROM im_devices", &res.Warnings) - res.PushTokenCount = s.queryScalarInt64(ctx, db, "im_device_push_tokens", "SELECT COUNT(*) FROM im_device_push_tokens", &res.Warnings) - res.InviteUseCount = s.queryScalarInt64(ctx, db, "im_registration_invite_uses", "SELECT COUNT(*) FROM im_registration_invite_uses", &res.Warnings) - res.OutboxPendingCount = s.queryScalarInt64(ctx, db, "im_business_events_outbox", "SELECT COUNT(*) FROM im_business_events_outbox WHERE status='pending'", &res.Warnings) + tenantId := tenant.Id + res.UserCount = s.queryScalarInt64(ctx, db, "im_users.total", "SELECT COUNT(*) FROM im_users WHERE tenant_id=?", &res.Warnings, tenantId) + res.ActiveUserCount = s.queryScalarInt64(ctx, db, "im_users.active", "SELECT COUNT(*) FROM im_users WHERE tenant_id=? AND state=0", &res.Warnings, tenantId) + res.GroupCount = s.queryScalarInt64(ctx, db, "im_group_settings", "SELECT COUNT(*) FROM im_group_settings WHERE tenant_id=? AND group_status<>'dissolved'", &res.Warnings, tenantId) + res.DeviceCount = s.queryScalarInt64(ctx, db, "im_devices", "SELECT COUNT(*) FROM im_devices WHERE tenant_id=?", &res.Warnings, tenantId) + res.PushTokenCount = s.queryScalarInt64(ctx, db, "im_device_push_tokens", "SELECT COUNT(*) FROM im_device_push_tokens WHERE tenant_id=?", &res.Warnings, tenantId) + res.InviteUseCount = s.queryScalarInt64(ctx, db, "im_registration_invite_uses", "SELECT COUNT(*) FROM im_registration_invite_uses WHERE tenant_id=?", &res.Warnings, tenantId) + res.OutboxPendingCount = s.queryScalarInt64(ctx, db, "im_business_events_outbox", "SELECT COUNT(*) FROM im_business_events_outbox WHERE tenant_id=? AND status='pending'", &res.Warnings, tenantId) return } -func (s *sSysImTenant) queryScalarInt64(ctx context.Context, db gdb.DB, label, sql string, warnings *[]string) int64 { - value, err := db.GetValue(ctx, sql) +func (s *sSysImTenant) queryScalarInt64(ctx context.Context, db gdb.DB, label, sql string, warnings *[]string, args ...any) int64 { + value, err := db.GetValue(ctx, sql, args...) if err != nil { *warnings = append(*warnings, label+": "+err.Error()) return 0 @@ -687,6 +715,9 @@ func (s *sSysImTenant) Restore(ctx context.Context, in *sysin.ImTenantLifecycleI if err = in.Filter(ctx); err != nil { return } + if err = provisionTenantMessageTables(ctx, in.Id); err != nil { + return gerror.Wrap(err, "恢复商户前创建业务分表失败") + } if _, err = s.Model(ctx). WherePri(in.Id). WhereIn(dao.ImTenant.Columns().State, []int{consts.ImTenantStateSuspended, consts.ImTenantStateDeleting}). diff --git a/server/internal/logic/sys/im_tenant_tables.go b/server/internal/logic/sys/im_tenant_tables.go new file mode 100644 index 0000000..f5fa8b0 --- /dev/null +++ b/server/internal/logic/sys/im_tenant_tables.go @@ -0,0 +1,131 @@ +package sys + +import ( + "context" + "fmt" + "hotgo/internal/library/imtenant" + "time" + + "github.com/gogf/gf/v2/database/gdb" + "github.com/gogf/gf/v2/errors/gerror" +) + +const imTenantIDSequenceTable = "im_tenant_id_sequence" + +func sharedDatabaseConfig(ctx context.Context) (dbName, dbDSN string, err error) { + row, err := imtenant.RegistryDB().GetOne(ctx, "SELECT db_name,db_dsn FROM im_tenant WHERE code='default' LIMIT 1") + if err != nil { + return "", "", gerror.Wrap(err, "获取共享业务库配置失败") + } + if row.IsEmpty() { + return "", "", gerror.New("默认商户不存在，无法确定共享业务库") + } + return row["db_name"].String(), row["db_dsn"].String(), nil +} + +func ensureTenantIDSequence(ctx context.Context) error { + db := imtenant.RegistryDB() + if _, err := db.Exec(ctx, `CREATE TABLE IF NOT EXISTS im_tenant_id_sequence ( + sequence_name VARCHAR(32) NOT NULL, + next_id BIGINT NOT NULL, + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY(sequence_name) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`); err != nil { + return gerror.Wrap(err, "初始化商户编号序列失败") + } + _, err := db.Exec(ctx, `INSERT INTO im_tenant_id_sequence(sequence_name,next_id) + SELECT 'tenant',COALESCE(MAX(id),0)+1 FROM im_tenant + ON DUPLICATE KEY UPDATE next_id=GREATEST(next_id,VALUES(next_id))`) + return gerror.Wrap(err, "校准商户编号序列失败") +} + +func nextTenantID(tx gdb.TX) (int64, error) { + value, err := tx.GetValue("SELECT next_id FROM " + imTenantIDSequenceTable + " WHERE sequence_name='tenant' FOR UPDATE") + if err != nil { + return 0, gerror.Wrap(err, "锁定商户编号序列失败") + } + id := value.Int64() + if id <= 0 { + return 0, gerror.New("商户编号序列不正确") + } + if _, err = tx.Exec("UPDATE "+imTenantIDSequenceTable+" SET next_id=? WHERE sequence_name='tenant'", id+1); err != nil { + return 0, gerror.Wrap(err, "更新商户编号序列失败") + } + return id, nil +} + +func tenantTablePrefix(tenantID int64) string { + return fmt.Sprintf("im_%03d_", tenantID) +} + +func provisionTenantMessageTables(ctx context.Context, tenantID int64) error { + if tenantID <= 0 { + return gerror.New("商户ID不能为空") + } + db, err := imtenant.SharedDB(ctx) + if err != nil { + return err + } + prefix := tenantTablePrefix(tenantID) + baseTables := [][2]string{ + {prefix + "friendships", "im_friendships"}, + {prefix + "messages", "im_messages"}, + } + 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 { + return gerror.Wrapf(err, "创建商户业务表 %s 失败", pair[0]) + } + if _, err = db.Exec(ctx, fmt.Sprintf("ALTER TABLE `%s` MODIFY tenant_id BIGINT NOT NULL DEFAULT %d COMMENT '商户ID'", pair[0], tenantID)); err != nil { + return gerror.Wrapf(err, "设置商户业务表 %s 默认商户失败", pair[0]) + } + } + + first := time.Date(time.Now().UTC().Year(), time.Now().UTC().Month(), 1, 0, 0, 0, 0, time.UTC) + for _, offset := range []int{-1, 0, 1, 2} { + start := first.AddDate(0, offset, 0) + month := start.Format("200601") + table := prefix + "group_messages_" + month + if err = createTenantGroupMessageTable(ctx, db, table, tenantID); err != nil { + return err + } + status := "prepared" + if offset == -1 { + status = "sealing" + } else if offset == 0 { + status = "writable" + } + if _, err = db.Exec(ctx, `INSERT INTO im_group_message_shards + (tenant_id,shard_month,table_name,shard_start,shard_end,status) + VALUES(?,?,?,?,?,?) ON DUPLICATE KEY UPDATE table_name=VALUES(table_name), + shard_start=VALUES(shard_start),shard_end=VALUES(shard_end), + status=IF(status IN ('sealed','archived'),status,VALUES(status))`, + tenantID, month, table, start, start.AddDate(0, 1, 0), status); err != nil { + return gerror.Wrapf(err, "登记商户群消息月表 %s 失败", table) + } + } + return nil +} + +func createTenantGroupMessageTable(ctx context.Context, db gdb.DB, table string, tenantID int64) error { + ddl := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s ( + tenant_id BIGINT NOT NULL DEFAULT %d, message_id BIGINT NOT NULL, topic VARCHAR(64) NOT NULL, group_id BIGINT NOT NULL, + seqid BIGINT NOT NULL, from_user BIGINT NOT NULL, client_msg_id VARCHAR(64), + msg_type VARCHAR(32) NOT NULL DEFAULT 'text', status VARCHAR(32) NOT NULL DEFAULT 'normal', + content JSON, content_nonce VARBINARY(12), key_version INT NOT NULL DEFAULT 0, + head JSON, mentioned_user_ids JSON, mention_all TINYINT(1) NOT NULL DEFAULT 0, + reply_seqid BIGINT, root_seqid BIGINT, file_count INT NOT NULL DEFAULT 0, + content_size INT NOT NULL DEFAULT 0, has_risk TINYINT(1) NOT NULL DEFAULT 0, + audit_status VARCHAR(32) NOT NULL DEFAULT 'pass', version INT NOT NULL DEFAULT 1, + recalled_at DATETIME(3), recalled_by BIGINT, deleted_at DATETIME(3), expires_at DATETIME(3), + created_at DATETIME(3) NOT NULL, updated_at DATETIME(3) NOT NULL, + PRIMARY KEY(message_id), UNIQUE INDEX uk_group_seq(group_id, seqid), + UNIQUE INDEX uk_topic_seq(topic, seqid), UNIQUE INDEX uk_sender_client(topic, from_user, client_msg_id), + INDEX idx_group_status_seq(group_id, status, seqid), INDEX idx_topic_created(topic, created_at), + INDEX idx_topic_from_seq(topic, from_user, seqid), INDEX idx_audit_created(audit_status, created_at), + INDEX idx_expires(expires_at) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, table, tenantID) + if _, err := db.Exec(ctx, ddl); err != nil { + return gerror.Wrapf(err, "创建商户群消息月表 %s 失败", table) + } + return nil +} diff --git a/server/internal/logic/sys/im_user.go b/server/internal/logic/sys/im_user.go index 293142c..f0b7260 100644 --- a/server/internal/logic/sys/im_user.go +++ b/server/internal/logic/sys/im_user.go @@ -861,7 +861,7 @@ func updateIMUserStatus(ctx context.Context, db gdb.DB, scope imAdminScope, user eventKeyBytes := sha256.Sum256([]byte(fmt.Sprintf("user-status:%d:%d:%d:%s", scope.TenantId, userId, state, requestId))) eventKey := hex.EncodeToString(eventKeyBytes[:]) return db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { - current, txErr := tx.GetValue("SELECT state FROM im_users WHERE id=? LIMIT 1", userId) + current, txErr := tx.GetValue("SELECT state FROM im_users WHERE tenant_id=? AND id=? LIMIT 1", scope.TenantId, userId) if txErr != nil { return txErr } @@ -874,17 +874,17 @@ func updateIMUserStatus(ctx context.Context, db gdb.DB, scope imAdminScope, user if current.Int() == state { return nil } - if _, txErr = tx.Exec("UPDATE im_users SET state=?, updatedat=? WHERE id=?", state, now, userId); txErr != nil { + if _, txErr = tx.Exec("UPDATE im_users SET state=?, updatedat=? WHERE tenant_id=? AND id=?", state, now, scope.TenantId, userId); txErr != nil { return txErr } if _, txErr = tx.Exec(`INSERT INTO im_user_security - (id,user_id,auth_version,created_at,updated_at) - VALUES(?,?,?,?,?) + (tenant_id,id,user_id,auth_version,created_at,updated_at) + VALUES(?,?,?,?,?,?) ON DUPLICATE KEY UPDATE auth_version=auth_version+1,updated_at=VALUES(updated_at)`, - securityId, userId, 2, now, now); txErr != nil { + scope.TenantId, securityId, userId, 2, now, now); txErr != nil { return txErr } - version, txErr := tx.GetValue("SELECT auth_version FROM im_user_security WHERE user_id=?", userId) + version, txErr := tx.GetValue("SELECT auth_version FROM im_user_security WHERE tenant_id=? AND user_id=?", scope.TenantId, userId) if txErr != nil { return txErr } @@ -892,9 +892,9 @@ func updateIMUserStatus(ctx context.Context, db gdb.DB, scope imAdminScope, user "user_id": userId, "operator_id": scope.OperatorId, "request_id": requestId, "state": state, "auth_version": version.Int(), }) _, txErr = tx.Exec(`INSERT INTO im_business_events_outbox - (id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at) - VALUES(?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?) - ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, eventId, eventKey, "user", + (tenant_id,id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?) + ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, scope.TenantId, eventId, eventKey, "user", fmt.Sprintf("%d", userId), 0, "security.user_status_changed", string(payload), now, now) return txErr }) @@ -920,7 +920,7 @@ func revokeIMUserSessions(ctx context.Context, db gdb.DB, scope imAdminScope, us eventKeyBytes := sha256.Sum256([]byte(fmt.Sprintf("%s:%d:%d:%s", eventType, scope.TenantId, userId, requestId))) eventKey := hex.EncodeToString(eventKeyBytes[:]) return db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { - exists, txErr := tx.GetValue("SELECT EXISTS(SELECT 1 FROM im_users WHERE id=? LIMIT 1)", userId) + exists, txErr := tx.GetValue("SELECT EXISTS(SELECT 1 FROM im_users WHERE tenant_id=? AND id=? LIMIT 1)", scope.TenantId, userId) if txErr != nil { return txErr } @@ -928,13 +928,13 @@ func revokeIMUserSessions(ctx context.Context, db gdb.DB, scope imAdminScope, us return gerror.New("IM用户不存在") } if _, txErr = tx.Exec(`INSERT INTO im_user_security - (id,user_id,auth_version,created_at,updated_at) - VALUES(?,?,?,?,?) + (tenant_id,id,user_id,auth_version,created_at,updated_at) + VALUES(?,?,?,?,?,?) ON DUPLICATE KEY UPDATE auth_version=auth_version+1,updated_at=VALUES(updated_at)`, - securityId, userId, 2, now, now); txErr != nil { + scope.TenantId, securityId, userId, 2, now, now); txErr != nil { return txErr } - version, txErr := tx.GetValue("SELECT auth_version FROM im_user_security WHERE user_id=?", userId) + version, txErr := tx.GetValue("SELECT auth_version FROM im_user_security WHERE tenant_id=? AND user_id=?", scope.TenantId, userId) if txErr != nil { return txErr } @@ -942,9 +942,9 @@ func revokeIMUserSessions(ctx context.Context, db gdb.DB, scope imAdminScope, us "user_id": userId, "operator_id": scope.OperatorId, "request_id": requestId, "auth_version": version.Int(), }) _, txErr = tx.Exec(`INSERT INTO im_business_events_outbox - (id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at) - VALUES(?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?) - ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, eventId, eventKey, "user", + (tenant_id,id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?) + ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, scope.TenantId, eventId, eventKey, "user", fmt.Sprintf("%d", userId), 0, eventType, string(payload), now, now) return txErr }) @@ -977,14 +977,14 @@ func resetIMUserPassword(ctx context.Context, db gdb.DB, scope imAdminScope, use "user_id": userId, "operator_id": scope.OperatorId, "request_id": requestId, }) return db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { - exists, txErr := tx.GetValue("SELECT EXISTS(SELECT 1 FROM im_users WHERE id=? LIMIT 1)", userId) + exists, txErr := tx.GetValue("SELECT EXISTS(SELECT 1 FROM im_users WHERE tenant_id=? AND id=? LIMIT 1)", scope.TenantId, userId) if txErr != nil { return txErr } if !exists.Bool() { return gerror.New("IM用户不存在") } - result, txErr := tx.Exec("UPDATE im_auth SET secret=? WHERE userid=?", passwordHash, userId) + result, txErr := tx.Exec("UPDATE im_auth SET secret=? WHERE tenant_id=? AND userid=?", passwordHash, scope.TenantId, userId) if txErr != nil { return txErr } @@ -992,14 +992,14 @@ func resetIMUserPassword(ctx context.Context, db gdb.DB, scope imAdminScope, use return gerror.New("用户认证记录不存在") } if _, txErr = tx.Exec(`INSERT INTO im_user_security - (id,user_id,auth_version,password_updated_at,created_at,updated_at) - VALUES(?,?,?,?,?,?) + (tenant_id,id,user_id,auth_version,password_updated_at,created_at,updated_at) + VALUES(?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE auth_version=auth_version+1, password_updated_at=VALUES(password_updated_at),updated_at=VALUES(updated_at)`, - securityId, userId, 2, now, now, now); txErr != nil { + scope.TenantId, securityId, userId, 2, now, now, now); txErr != nil { return txErr } - version, txErr := tx.GetValue("SELECT auth_version FROM im_user_security WHERE user_id=?", userId) + version, txErr := tx.GetValue("SELECT auth_version FROM im_user_security WHERE tenant_id=? AND user_id=?", scope.TenantId, userId) if txErr != nil { return txErr } @@ -1008,9 +1008,9 @@ func resetIMUserPassword(ctx context.Context, db gdb.DB, scope imAdminScope, use eventPayload["auth_version"] = version.Int() payload, _ = json.Marshal(eventPayload) _, txErr = tx.Exec(`INSERT INTO im_business_events_outbox - (id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at) - VALUES(?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?) - ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, eventId, eventKey, "user", + (tenant_id,id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?) + ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, scope.TenantId, eventId, eventKey, "user", fmt.Sprintf("%d", userId), 0, "security.password_reset", string(payload), now, now) return txErr }) diff --git a/server/internal/model/input/sysin/im_tenant.go b/server/internal/model/input/sysin/im_tenant.go index 2fe67f0..f4af04a 100644 --- a/server/internal/model/input/sysin/im_tenant.go +++ b/server/internal/model/input/sysin/im_tenant.go @@ -20,6 +20,7 @@ import ( // ImTenantInsertFields 新增IM租户字段过滤 type ImTenantInsertFields struct { + Id int64 `json:"id" dc:"商户ID"` Code string `json:"code" dc:"租户编码"` Name string `json:"name" dc:"租户名称"` TenantDesc string `json:"tenantDesc" dc:"租户描述"`