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

feat: support shared tenant database sharding

Browse Source
This commit is contained in:
richjohnrambo
2026-09-22 15:21:34 +08:00
parent 0596599db9
commit 851eca97ed
6 changed files with 265 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/library/imtenant/db.go
+31 -10
View File
@@ -23,7 +23,7 @@ const registryDbGroup = "imRegistry"
var registryDbGroupOnce sync.Once 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) { func DB(ctx context.Context, tenantId int64) (db gdb.DB, tenant *entity.ImTenant, err error) {
if tenantId <= 0 { if tenantId <= 0 {
return nil, nil, gerror.New("租户ID不能为空") 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("租户不存在") return nil, nil, gerror.New("租户不存在")
} }
link, err := tenantLink(ctx, tenant) db, err = SharedDB(ctx)
if err != nil { if err != nil {
return nil, nil, err 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 { if cached, ok := tenantDbGroups.Load(group); ok && cached == link {
db, err = gdb.Instance(group) return gdb.Instance(group)
return db, tenant, err
} }
if err = gdb.SetConfigGroup(group, gdb.ConfigGroup{{Link: link}}); err != nil { if err := gdb.SetConfigGroup(group, gdb.ConfigGroup{{Link: link}}); err != nil {
return nil, nil, gerror.Wrap(err, "初始化租户业务库连接失败") return nil, gerror.Wrap(err, "初始化共享业务库连接失败")
} }
tenantDbGroups.Store(group, link) tenantDbGroups.Store(group, link)
db, err = gdb.Instance(group) db, err := gdb.Instance(group)
if err != nil { 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 // RegistryDB returns the dedicated tenant registry connection, or the default
server/internal/logic/sys/im_group.go
+26 -21
View File
@@ -52,7 +52,7 @@ func (s *sSysImGroup) List(ctx context.Context, in *sysin.ImGroupListInp) (list
return return
} }
mod := s.groupModel(ctx, db) mod := s.groupModel(ctx, db, tenant.Id)
mod = s.applyGroupWhere(mod, in) mod = s.applyGroupWhere(mod, in)
mod = mod.Page(in.Page, in.PerPage).OrderDesc("t.touchedat").OrderDesc("t.id") 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 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). return db.Model("im_topics t").Safe().Ctx(ctx).
LeftJoin("im_group_settings gs", "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.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.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)").
Fields( Fields(
"t.name topic", "t.name topic",
"t.state", "t.state",
@@ -103,6 +103,7 @@ func (s *sSysImGroup) groupModel(ctx context.Context, db gdb.DB) *gdb.Model {
"COALESCE(gs.version, 0) settings_version", "COALESCE(gs.version, 0) settings_version",
"gs.updated_at settings_updated_at", "gs.updated_at settings_updated_at",
). ).
Where("t.tenant_id", tenantId).
WhereLike("t.name", "grp%") 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). err = db.Model("im_topics t").Safe().Ctx(ctx).
LeftJoin("im_group_settings gs", "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.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.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)").
Fields( Fields(
"t.name topic", "t.name topic",
"t.state", "t.state",
@@ -176,6 +177,7 @@ func (s *sSysImGroup) View(ctx context.Context, in *sysin.ImGroupViewInp) (res *
"COALESCE(gs.version, 0) settings_version", "COALESCE(gs.version, 0) settings_version",
"gs.updated_at settings_updated_at", "gs.updated_at settings_updated_at",
). ).
Where("t.tenant_id", tenant.Id).
Where("t.name", in.Topic). Where("t.name", in.Topic).
WhereLike("t.name", "grp%"). WhereLike("t.name", "grp%").
Scan(&res) Scan(&res)
@@ -196,27 +198,27 @@ func (s *sSysImGroup) MemberList(ctx context.Context, in *sysin.ImGroupMemberLis
return return
} }
db, _, err := resolveImTenantDB(ctx, in.TenantId) db, tenant, err := resolveImTenantDB(ctx, in.TenantId)
if err != nil { if err != nil {
return return
} }
if s.hasGroupMemberTruth(ctx, db, in.Topic) { if s.hasGroupMemberTruth(ctx, db, tenant.Id, in.Topic) {
return s.memberListFromBusiness(ctx, db, in) 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 { 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("topic", topic).Count() count, err := db.Model("im_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, 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("im_group_members gm").Safe().Ctx(ctx).
LeftJoin("im_users u", "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.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.topic = gm.topic AND s.userid = 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( Fields(
"gm.user_id", "gm.user_id",
"up.nickname", "up.nickname",
@@ -242,6 +244,7 @@ func (s *sSysImGroup) memberListFromBusiness(ctx context.Context, db gdb.DB, in
"gm.removed_by", "gm.removed_by",
"gm.version", "gm.version",
). ).
Where("gm.tenant_id", tenantId).
Where("gm.topic", in.Topic) Where("gm.topic", in.Topic)
if in.UserId > 0 { if in.UserId > 0 {
mod = mod.Where("gm.user_id", in.UserId) mod = mod.Where("gm.user_id", in.UserId)
@@ -260,10 +263,10 @@ func (s *sSysImGroup) memberListFromBusiness(ctx context.Context, db gdb.DB, in
return 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). mod := db.Model("im_subscriptions s").Safe().Ctx(ctx).
LeftJoin("im_users u", "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.user_id = s.userid"). LeftJoin("im_user_profiles up", "up.tenant_id = s.tenant_id AND up.user_id = s.userid").
Fields( Fields(
"s.userid user_id", "s.userid user_id",
"up.nickname", "up.nickname",
@@ -289,6 +292,7 @@ func (s *sSysImGroup) memberListFromProjection(ctx context.Context, db gdb.DB, i
"0 removed_by", "0 removed_by",
"0 version", "0 version",
). ).
Where("s.tenant_id", tenantId).
Where("s.topic", in.Topic) Where("s.topic", in.Topic)
if in.UserId > 0 { if in.UserId > 0 {
@@ -335,6 +339,7 @@ func (s *sSysImGroup) Governance(ctx context.Context, in *sysin.ImGroupTopicInp)
"created_at", "created_at",
"updated_at", "updated_at",
). ).
Where("tenant_id", tenant.Id).
Where("topic", in.Topic). Where("topic", in.Topic).
Scan(&res) Scan(&res)
if err != nil { if err != nil {
server/internal/logic/sys/im_tenant.go
+50 -19
View File
@@ -222,13 +222,23 @@ func (s *sSysImTenant) saveTenant(ctx context.Context, in *sysin.ImTenantEditInp
super := isSuperTenantAdmin(ctx) super := isSuperTenantAdmin(ctx)
isCreate := in.Id == 0 isCreate := in.Id == 0
tenantID := in.Id tenantID := in.Id
requestedState := in.State
// 业务库名与独立连接串仅超管可维护，非超管提交一律强制忽略 if isCreate {
if !super { 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.DbName = ""
in.DbDsn = "" in.DbDsn = ""
} }
err = imtenant.RegistryDB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) (err error) { 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 { if err = s.checkCodeUnique(ctx, in.Id, in.Code); err != nil {
return return
@@ -255,16 +265,19 @@ func (s *sSysImTenant) saveTenant(ctx context.Context, in *sysin.ImTenantEditInp
return s.saveConfig(ctx, in, userId) return s.saveConfig(ctx, in, userId)
} }
in.Id, err = nextTenantID(tx)
if err != nil {
return err
}
in.CreatedBy = userId in.CreatedBy = userId
in.UpdatedBy = 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{}). Fields(sysin.ImTenantInsertFields{}).
Data(in).InsertAndGetId() Data(in).Insert()
if err != nil { if err != nil {
return gerror.Wrap(err, "新增IM租户失败，请稍后重试！") return gerror.Wrap(err, "新增IM租户失败，请稍后重试！")
} }
tenantID = id tenantID = in.Id
in.Id = id
return s.saveConfig(ctx, in, userId) return s.saveConfig(ctx, in, userId)
}) })
if err != nil { if err != nil {
@@ -277,13 +290,22 @@ func (s *sSysImTenant) saveTenant(ctx context.Context, in *sysin.ImTenantEditInp
s.rollbackTenantCreation(ctx, tenantID) s.rollbackTenantCreation(ctx, tenantID)
return err 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 != "" { } else if in.AdminPassword != "" {
// 编辑商户时填写了新密码则重置管理员密码，留空不修改 // 编辑商户时填写了新密码则重置管理员密码，留空不修改。
if err = s.resetTenantAdminPassword(ctx, tenantID, in.AdminPassword); err != nil { if err = s.resetTenantAdminPassword(ctx, tenantID, in.AdminPassword); err != nil {
return err return err
} }
} }
_ = publishImPolicyEvent(ctx, tenantID, in.Code, "tenant.features.updated", "moderation", 0) _ = publishImPolicyEvent(ctx, tenantID, in.Code, "tenant.features.updated", "moderation", 0)
return nil return nil
} }
@@ -550,6 +572,11 @@ func (s *sSysImTenant) Status(ctx context.Context, in *sysin.ImTenantStatusInp)
if err = in.Filter(ctx); err != nil { if err = in.Filter(ctx); err != nil {
return 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{ if _, err = s.Model(ctx).WherePri(in.Id).Data(g.Map{
dao.ImTenant.Columns().State: in.State, dao.ImTenant.Columns().State: in.State,
dao.ImTenant.Columns().UpdatedBy: contexts.GetUserId(ctx), 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.DbReachable = true
res.SchemaVersion = s.queryScalarString(ctx, db, "kvmeta.version", "SELECT value FROM kvmeta WHERE key_='version' LIMIT 1", &res.Warnings) 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) tenantId := tenant.Id
res.ActiveUserCount = s.queryScalarInt64(ctx, db, "im_users.active", "SELECT COUNT(*) FROM im_users WHERE state=0", &res.Warnings) res.UserCount = s.queryScalarInt64(ctx, db, "im_users.total", "SELECT COUNT(*) FROM im_users WHERE tenant_id=?", &res.Warnings, tenantId)
res.GroupCount = s.queryScalarInt64(ctx, db, "im_group_settings", "SELECT COUNT(*) FROM im_group_settings WHERE group_status<>'dissolved'", &res.Warnings) res.ActiveUserCount = s.queryScalarInt64(ctx, db, "im_users.active", "SELECT COUNT(*) FROM im_users WHERE tenant_id=? AND state=0", &res.Warnings, tenantId)
res.DeviceCount = s.queryScalarInt64(ctx, db, "im_devices", "SELECT COUNT(*) FROM im_devices", &res.Warnings) 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.PushTokenCount = s.queryScalarInt64(ctx, db, "im_device_push_tokens", "SELECT COUNT(*) FROM im_device_push_tokens", &res.Warnings) res.DeviceCount = s.queryScalarInt64(ctx, db, "im_devices", "SELECT COUNT(*) FROM im_devices WHERE tenant_id=?", &res.Warnings, tenantId)
res.InviteUseCount = s.queryScalarInt64(ctx, db, "im_registration_invite_uses", "SELECT COUNT(*) FROM im_registration_invite_uses", &res.Warnings) res.PushTokenCount = s.queryScalarInt64(ctx, db, "im_device_push_tokens", "SELECT COUNT(*) FROM im_device_push_tokens WHERE tenant_id=?", &res.Warnings, tenantId)
res.OutboxPendingCount = s.queryScalarInt64(ctx, db, "im_business_events_outbox", "SELECT COUNT(*) FROM im_business_events_outbox WHERE status='pending'", &res.Warnings) 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 return
} }
func (s *sSysImTenant) queryScalarInt64(ctx context.Context, db gdb.DB, label, sql string, warnings *[]string) int64 { func (s *sSysImTenant) queryScalarInt64(ctx context.Context, db gdb.DB, label, sql string, warnings *[]string, args ...any) int64 {
value, err := db.GetValue(ctx, sql) value, err := db.GetValue(ctx, sql, args...)
if err != nil { if err != nil {
*warnings = append(*warnings, label+": "+err.Error()) *warnings = append(*warnings, label+": "+err.Error())
return 0 return 0
@@ -687,6 +715,9 @@ func (s *sSysImTenant) Restore(ctx context.Context, in *sysin.ImTenantLifecycleI
if err = in.Filter(ctx); err != nil { if err = in.Filter(ctx); err != nil {
return return
} }
if err = provisionTenantMessageTables(ctx, in.Id); err != nil {
return gerror.Wrap(err, "恢复商户前创建业务分表失败")
}
if _, err = s.Model(ctx). if _, err = s.Model(ctx).
WherePri(in.Id). WherePri(in.Id).
WhereIn(dao.ImTenant.Columns().State, []int{consts.ImTenantStateSuspended, consts.ImTenantStateDeleting}). WhereIn(dao.ImTenant.Columns().State, []int{consts.ImTenantStateSuspended, consts.ImTenantStateDeleting}).
server/internal/logic/sys/im_tenant_tables.go
+131
View File
@@ -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
}
server/internal/logic/sys/im_user.go
+26 -26
View File
@@ -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))) eventKeyBytes := sha256.Sum256([]byte(fmt.Sprintf("user-status:%d:%d:%d:%s", scope.TenantId, userId, state, requestId)))
eventKey := hex.EncodeToString(eventKeyBytes[:]) eventKey := hex.EncodeToString(eventKeyBytes[:])
return db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { 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 { if txErr != nil {
return txErr return txErr
} }
@@ -874,17 +874,17 @@ func updateIMUserStatus(ctx context.Context, db gdb.DB, scope imAdminScope, user
if current.Int() == state { if current.Int() == state {
return nil 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 return txErr
} }
if _, txErr = tx.Exec(`INSERT INTO im_user_security if _, txErr = tx.Exec(`INSERT INTO im_user_security
(id,user_id,auth_version,created_at,updated_at) (tenant_id,id,user_id,auth_version,created_at,updated_at)
VALUES(?,?,?,?,?) VALUES(?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE auth_version=auth_version+1,updated_at=VALUES(updated_at)`, 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 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 { if txErr != nil {
return txErr 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(), "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 _, 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) (tenant_id,id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at)
VALUES(?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?) VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?)
ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, eventId, eventKey, "user", 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) fmt.Sprintf("%d", userId), 0, "security.user_status_changed", string(payload), now, now)
return txErr 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))) eventKeyBytes := sha256.Sum256([]byte(fmt.Sprintf("%s:%d:%d:%s", eventType, scope.TenantId, userId, requestId)))
eventKey := hex.EncodeToString(eventKeyBytes[:]) eventKey := hex.EncodeToString(eventKeyBytes[:])
return db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { 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 { if txErr != nil {
return txErr return txErr
} }
@@ -928,13 +928,13 @@ func revokeIMUserSessions(ctx context.Context, db gdb.DB, scope imAdminScope, us
return gerror.New("IM用户不存在") return gerror.New("IM用户不存在")
} }
if _, txErr = tx.Exec(`INSERT INTO im_user_security if _, txErr = tx.Exec(`INSERT INTO im_user_security
(id,user_id,auth_version,created_at,updated_at) (tenant_id,id,user_id,auth_version,created_at,updated_at)
VALUES(?,?,?,?,?) VALUES(?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE auth_version=auth_version+1,updated_at=VALUES(updated_at)`, 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 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 { if txErr != nil {
return txErr 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(), "user_id": userId, "operator_id": scope.OperatorId, "request_id": requestId, "auth_version": version.Int(),
}) })
_, txErr = tx.Exec(`INSERT INTO im_business_events_outbox _, 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) (tenant_id,id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at)
VALUES(?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?) VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?)
ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, eventId, eventKey, "user", ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, scope.TenantId, eventId, eventKey, "user",
fmt.Sprintf("%d", userId), 0, eventType, string(payload), now, now) fmt.Sprintf("%d", userId), 0, eventType, string(payload), now, now)
return txErr 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, "user_id": userId, "operator_id": scope.OperatorId, "request_id": requestId,
}) })
return db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { 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 { if txErr != nil {
return txErr return txErr
} }
if !exists.Bool() { if !exists.Bool() {
return gerror.New("IM用户不存在") 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 { if txErr != nil {
return txErr return txErr
} }
@@ -992,14 +992,14 @@ func resetIMUserPassword(ctx context.Context, db gdb.DB, scope imAdminScope, use
return gerror.New("用户认证记录不存在") return gerror.New("用户认证记录不存在")
} }
if _, txErr = tx.Exec(`INSERT INTO im_user_security if _, txErr = tx.Exec(`INSERT INTO im_user_security
(id,user_id,auth_version,password_updated_at,created_at,updated_at) (tenant_id,id,user_id,auth_version,password_updated_at,created_at,updated_at)
VALUES(?,?,?,?,?,?) VALUES(?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE auth_version=auth_version+1, ON DUPLICATE KEY UPDATE auth_version=auth_version+1,
password_updated_at=VALUES(password_updated_at),updated_at=VALUES(updated_at)`, 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 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 { if txErr != nil {
return txErr return txErr
} }
@@ -1008,9 +1008,9 @@ func resetIMUserPassword(ctx context.Context, db gdb.DB, scope imAdminScope, use
eventPayload["auth_version"] = version.Int() eventPayload["auth_version"] = version.Int()
payload, _ = json.Marshal(eventPayload) payload, _ = json.Marshal(eventPayload)
_, txErr = tx.Exec(`INSERT INTO im_business_events_outbox _, 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) (tenant_id,id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at)
VALUES(?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?) VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?)
ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, eventId, eventKey, "user", 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) fmt.Sprintf("%d", userId), 0, "security.password_reset", string(payload), now, now)
return txErr return txErr
}) })
server/internal/model/input/sysin/im_tenant.go
+1
View File
@@ -20,6 +20,7 @@ import (
// ImTenantInsertFields 新增IM租户字段过滤 // ImTenantInsertFields 新增IM租户字段过滤
type ImTenantInsertFields struct { type ImTenantInsertFields struct {
Id int64 `json:"id" dc:"商户ID"`
Code string `json:"code" dc:"租户编码"` Code string `json:"code" dc:"租户编码"`
Name string `json:"name" dc:"租户名称"` Name string `json:"name" dc:"租户名称"`
TenantDesc string `json:"tenantDesc" dc:"租户描述"` TenantDesc string `json:"tenantDesc" dc:"租户描述"`