feat: add tenant-aware IM management console
This commit is contained in:
richjohnrambo
@@ -78,6 +78,15 @@ func GetUserId(ctx context.Context) int64 {
|
||||
return user.Id
|
||||
}
|
||||
|
||||
// GetTenantId 获取当前身份关联的统一租户ID（im_tenant.id）。
|
||||
func GetTenantId(ctx context.Context) int64 {
|
||||
user := GetUser(ctx)
|
||||
if user == nil {
|
||||
return 0
|
||||
}
|
||||
return user.TenantId
|
||||
}
|
||||
|
||||
// GetRoleId 获取用户角色ID
|
||||
func GetRoleId(ctx context.Context) int64 {
|
||||
user := GetUser(ctx)
|
||||
|
||||
@@ -0,0 +1,147 @@
|
||||
// Package imtenant provides helpers for accessing tenant business databases.
|
||||
package imtenant
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"hash/fnv"
|
||||
"hotgo/internal/dao"
|
||||
"hotgo/internal/model/entity"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
_ "github.com/gogf/gf/contrib/drivers/mysql/v2"
|
||||
"github.com/gogf/gf/v2/database/gdb"
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
)
|
||||
|
||||
var tenantDbGroups sync.Map
|
||||
|
||||
const registryDbGroup = "imRegistry"
|
||||
|
||||
// DB resolves a tenant registry record and returns the corresponding 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不能为空")
|
||||
}
|
||||
|
||||
if err = RegistryDB().Model(dao.ImTenant.Table()).Safe().Ctx(ctx).WherePri(tenantId).Scan(&tenant); err != nil {
|
||||
return nil, nil, gerror.Wrap(err, "获取租户信息失败")
|
||||
}
|
||||
if tenant == nil || tenant.Id <= 0 {
|
||||
return nil, nil, gerror.New("租户不存在")
|
||||
}
|
||||
|
||||
link, err := tenantLink(ctx, tenant)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
group := fmt.Sprintf("im_tenant_%d_%x", tenant.Id, hash(link))
|
||||
if cached, ok := tenantDbGroups.Load(group); ok && cached == link {
|
||||
db, err = gdb.Instance(group)
|
||||
return db, tenant, err
|
||||
}
|
||||
|
||||
if err = gdb.SetConfigGroup(group, gdb.ConfigGroup{{Link: link}}); err != nil {
|
||||
return nil, nil, gerror.Wrap(err, "初始化租户业务库连接失败")
|
||||
}
|
||||
tenantDbGroups.Store(group, link)
|
||||
|
||||
db, err = gdb.Instance(group)
|
||||
if err != nil {
|
||||
return nil, nil, gerror.Wrap(err, "获取租户业务库连接失败")
|
||||
}
|
||||
return db, tenant, nil
|
||||
}
|
||||
|
||||
// RegistryDB returns the dedicated tenant registry connection, or the default
|
||||
// HotGo database when a separate registry group is not configured.
|
||||
func RegistryDB() gdb.DB {
|
||||
if configs, err := gdb.GetConfigGroup(registryDbGroup); err == nil && len(configs) > 0 {
|
||||
return g.DB(registryDbGroup)
|
||||
}
|
||||
return g.DB()
|
||||
}
|
||||
|
||||
func tenantLink(ctx context.Context, tenant *entity.ImTenant) (string, error) {
|
||||
if tenant.DbDsn != "" {
|
||||
return normalizeMySQLTimeOptions(tenant.DbDsn)
|
||||
}
|
||||
if tenant.DbName == "" {
|
||||
return "", gerror.New("租户业务库未配置")
|
||||
}
|
||||
|
||||
defaultLink := defaultDbLink(ctx)
|
||||
if defaultLink == "" {
|
||||
return "", gerror.New("默认数据库连接未配置，无法推导租户业务库连接")
|
||||
}
|
||||
|
||||
link, err := replaceMySQLDatabase(defaultLink, tenant.DbName)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return link, nil
|
||||
}
|
||||
|
||||
func defaultDbLink(ctx context.Context) string {
|
||||
if configs, err := gdb.GetConfigGroup("default"); err == nil && len(configs) > 0 {
|
||||
if configs[0].Link != "" {
|
||||
return configs[0].Link
|
||||
}
|
||||
}
|
||||
|
||||
cfg := g.Cfg()
|
||||
if !cfg.Available(ctx) {
|
||||
return ""
|
||||
}
|
||||
defaultLink := cfg.MustGet(ctx, "database.default.0.link").String()
|
||||
if defaultLink != "" {
|
||||
return defaultLink
|
||||
}
|
||||
return cfg.MustGet(ctx, "database.default.link").String()
|
||||
}
|
||||
|
||||
func replaceMySQLDatabase(link string, dbName string) (string, error) {
|
||||
if !strings.HasPrefix(link, "mysql:") {
|
||||
return "", gerror.New("仅支持从 MySQL 默认连接推导租户业务库连接")
|
||||
}
|
||||
|
||||
dbStart := strings.LastIndex(link, ")/")
|
||||
if dbStart < 0 {
|
||||
return "", gerror.New("默认数据库连接格式不正确")
|
||||
}
|
||||
dbStart += 2
|
||||
queryStart := strings.Index(link[dbStart:], "?")
|
||||
if queryStart < 0 {
|
||||
return normalizeMySQLTimeOptions(link[:dbStart] + dbName)
|
||||
}
|
||||
queryStart += dbStart
|
||||
return normalizeMySQLTimeOptions(link[:dbStart] + dbName + link[queryStart:])
|
||||
}
|
||||
|
||||
func normalizeMySQLTimeOptions(link string) (string, error) {
|
||||
if !strings.HasPrefix(link, "mysql:") {
|
||||
return "", gerror.New("租户业务库仅支持 MySQL")
|
||||
}
|
||||
parts := strings.SplitN(link, "?", 2)
|
||||
values := make(url.Values)
|
||||
if len(parts) == 2 {
|
||||
var err error
|
||||
values, err = url.ParseQuery(parts[1])
|
||||
if err != nil {
|
||||
return "", gerror.Wrap(err, "租户业务库连接参数格式不正确")
|
||||
}
|
||||
}
|
||||
values.Set("loc", "UTC")
|
||||
values.Set("parseTime", "true")
|
||||
return parts[0] + "?" + values.Encode(), nil
|
||||
}
|
||||
|
||||
func hash(s string) uint64 {
|
||||
h := fnv.New64a()
|
||||
_, _ = h.Write([]byte(s))
|
||||
return h.Sum64()
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
package imtenant
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestReplaceMySQLDatabaseUsesUTC(t *testing.T) {
|
||||
link, err := replaceMySQLDatabase("mysql:user:pass@tcp(127.0.0.1:3306)/hotgo?loc=Local&parseTime=false&charset=utf8mb4", "tinode")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, expected := range []string{"/tinode?", "loc=UTC", "parseTime=true", "charset=utf8mb4"} {
|
||||
if !strings.Contains(link, expected) {
|
||||
t.Fatalf("link %q does not contain %q", link, expected)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNextSnowflakeID(t *testing.T) {
|
||||
first, err := NextSnowflakeID(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
second, err := NextSnowflakeID(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if first <= 0 || second <= first {
|
||||
t.Fatalf("Snowflake IDs are not increasing: %d, %d", first, second)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
package imtenant
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
"sync"
|
||||
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
sf "github.com/tinode/snowflake"
|
||||
)
|
||||
|
||||
const defaultSnowflakeWorkerID = 1023
|
||||
|
||||
var (
|
||||
snowflakeOnce sync.Once
|
||||
snowflakeGen *sf.SnowFlake
|
||||
snowflakeErr error
|
||||
)
|
||||
|
||||
// NextSnowflakeID returns an unencrypted Snowflake ID suitable for IM BIGINT keys.
|
||||
func NextSnowflakeID(ctx context.Context) (int64, error) {
|
||||
snowflakeOnce.Do(func() {
|
||||
workerID := g.Cfg().MustGet(ctx, "imDirectDb.snowflakeWorkerId", defaultSnowflakeWorkerID).Int()
|
||||
if value := os.Getenv("IM_DIRECT_DB_SNOWFLAKE_WORKER_ID"); value != "" {
|
||||
workerID, snowflakeErr = strconv.Atoi(value)
|
||||
if snowflakeErr != nil {
|
||||
snowflakeErr = fmt.Errorf("invalid IM_DIRECT_DB_SNOWFLAKE_WORKER_ID: %w", snowflakeErr)
|
||||
return
|
||||
}
|
||||
}
|
||||
if workerID < 0 || workerID > 1023 {
|
||||
snowflakeErr = fmt.Errorf("IM Snowflake worker ID must be between 0 and 1023")
|
||||
return
|
||||
}
|
||||
snowflakeGen, snowflakeErr = sf.NewSnowFlake(uint32(workerID))
|
||||
})
|
||||
if snowflakeErr != nil {
|
||||
return 0, snowflakeErr
|
||||
}
|
||||
id, err := snowflakeGen.Next()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if id == 0 || id > uint64(^uint64(0)>>1) {
|
||||
return 0, fmt.Errorf("generated invalid IM Snowflake ID")
|
||||
}
|
||||
return int64(id), nil
|
||||
}
|
||||
Reference in New Issue
Block a user