fix: route workbench requests to tenant database
This commit is contained in:
+121
-4
@@ -7,19 +7,32 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.118im.com/im_team/im_workbench/internal/config"
|
||||
_ "github.com/go-sql-driver/mysql"
|
||||
mysqlDriver "github.com/go-sql-driver/mysql"
|
||||
"github.com/jmoiron/sqlx"
|
||||
)
|
||||
|
||||
type Store struct {
|
||||
db *sqlx.DB
|
||||
db *sqlx.DB
|
||||
dsn string
|
||||
tenantStores map[int64]*Store
|
||||
tenantStoreMu sync.Mutex
|
||||
}
|
||||
|
||||
func Open(cfg config.MySQLConfig) (*Store, error) {
|
||||
db, err := sqlx.Open("mysql", cfg.DSN)
|
||||
st, err := openDSN(cfg.DSN)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
st.tenantStores = make(map[int64]*Store)
|
||||
return st, nil
|
||||
}
|
||||
|
||||
func openDSN(dsn string) (*Store, error) {
|
||||
db, err := sqlx.Open("mysql", dsn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -32,16 +45,120 @@ func Open(cfg config.MySQLConfig) (*Store, error) {
|
||||
_ = db.Close()
|
||||
return nil, err
|
||||
}
|
||||
return &Store{db: db}, nil
|
||||
return &Store{db: db, dsn: dsn}, nil
|
||||
}
|
||||
|
||||
func (s *Store) Close() error {
|
||||
if s == nil || s.db == nil {
|
||||
return nil
|
||||
}
|
||||
s.tenantStoreMu.Lock()
|
||||
for id, st := range s.tenantStores {
|
||||
if st != nil && st != s {
|
||||
_ = st.Close()
|
||||
}
|
||||
delete(s.tenantStores, id)
|
||||
}
|
||||
s.tenantStoreMu.Unlock()
|
||||
return s.db.Close()
|
||||
}
|
||||
|
||||
type tenantDBConfig struct {
|
||||
ID int64 `db:"id"`
|
||||
Code string `db:"code"`
|
||||
DBName string `db:"db_name"`
|
||||
DBDSN string `db:"db_dsn"`
|
||||
State int `db:"state"`
|
||||
}
|
||||
|
||||
// ForTenant returns a Store connected to the tenant database configured by manager/hotgo.
|
||||
func (s *Store) ForTenant(ctx context.Context, tenantID int64, tenantCode string) (*Store, error) {
|
||||
if s == nil {
|
||||
return nil, errors.New("store is nil")
|
||||
}
|
||||
if tenantID <= 0 && strings.TrimSpace(tenantCode) == "" {
|
||||
return s, nil
|
||||
}
|
||||
cfg, err := s.tenantDBConfig(ctx, tenantID, tenantCode)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if cfg.State != 1 {
|
||||
return nil, fmt.Errorf("tenant disabled: id=%d code=%s", cfg.ID, cfg.Code)
|
||||
}
|
||||
dsn, err := tenantDSN(s.dsn, cfg.DBName, cfg.DBDSN)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if dsn == "" || dsn == s.dsn {
|
||||
return s, nil
|
||||
}
|
||||
cacheID := cfg.ID
|
||||
if cacheID <= 0 {
|
||||
cacheID = tenantID
|
||||
}
|
||||
s.tenantStoreMu.Lock()
|
||||
if st := s.tenantStores[cacheID]; st != nil {
|
||||
s.tenantStoreMu.Unlock()
|
||||
return st, nil
|
||||
}
|
||||
s.tenantStoreMu.Unlock()
|
||||
|
||||
st, err := openDSN(dsn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s.tenantStoreMu.Lock()
|
||||
if existing := s.tenantStores[cacheID]; existing != nil {
|
||||
s.tenantStoreMu.Unlock()
|
||||
_ = st.Close()
|
||||
return existing, nil
|
||||
}
|
||||
s.tenantStores[cacheID] = st
|
||||
s.tenantStoreMu.Unlock()
|
||||
return st, nil
|
||||
}
|
||||
|
||||
func (s *Store) tenantDBConfig(ctx context.Context, tenantID int64, tenantCode string) (*tenantDBConfig, error) {
|
||||
var cfg tenantDBConfig
|
||||
if tenantID > 0 {
|
||||
err := s.db.GetContext(ctx, &cfg, `SELECT id,code,IFNULL(db_name,'') AS db_name,IFNULL(db_dsn,'') AS db_dsn,state
|
||||
FROM hotgo.im_tenant WHERE id=? LIMIT 1`, tenantID)
|
||||
if err == nil {
|
||||
return &cfg, nil
|
||||
}
|
||||
if !errors.Is(err, sql.ErrNoRows) || strings.TrimSpace(tenantCode) == "" {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
err := s.db.GetContext(ctx, &cfg, `SELECT id,code,IFNULL(db_name,'') AS db_name,IFNULL(db_dsn,'') AS db_dsn,state
|
||||
FROM hotgo.im_tenant WHERE code=? LIMIT 1`, strings.TrimSpace(tenantCode))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &cfg, nil
|
||||
}
|
||||
|
||||
func tenantDSN(baseDSN, dbName, dbDSN string) (string, error) {
|
||||
dbDSN = strings.TrimSpace(dbDSN)
|
||||
if strings.HasPrefix(dbDSN, "mysql:") {
|
||||
dbDSN = strings.TrimPrefix(dbDSN, "mysql:")
|
||||
}
|
||||
if dbDSN != "" {
|
||||
return dbDSN, nil
|
||||
}
|
||||
dbName = strings.TrimSpace(dbName)
|
||||
if dbName == "" {
|
||||
return baseDSN, nil
|
||||
}
|
||||
cfg, err := mysqlDriver.ParseDSN(baseDSN)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
cfg.DBName = dbName
|
||||
return cfg.FormatDSN(), nil
|
||||
}
|
||||
|
||||
func (s *Store) WorkbenchModules(ctx context.Context, userID int64) ([]WorkbenchModule, int, error) {
|
||||
var modules []WorkbenchModule
|
||||
err := s.db.SelectContext(ctx, &modules, `SELECT code,name,IFNULL(icon_file_id,'') AS icon_file_id,route,sort_order,0 AS unread_count,enabled,permission
|
||||
|
||||
Reference in New Issue
Block a user