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

feat: split workbench core mysql access

Browse Source
This commit is contained in:
Your Name
2026-10-06 23:44:37 +08:00
parent 71b49ddcf7
commit 8d32376de8
6 changed files with 245 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/app/app.go
+1 -1
View File
@@ -33,7 +33,7 @@ func New(cfg *config.Config) (*App, error) {
return nil, err
}
cfg.AI = managerconfig.ResolveAIConfig(context.Background(), cfg.Manager, cfg.AI)
st, err := store.Open(cfg.MySQL)
st, err := store.Open(cfg.MySQL, cfg.CoreMySQL)
if err != nil {
return nil, err
}
internal/config/config.go
+15
View File
@@ -15,6 +15,7 @@ import (
type Config struct {
Listen string `json:"listen"`
MySQL MySQLConfig `json:"mysql"`
CoreMySQL MySQLConfig `json:"core_mysql"`
Redis RedisConfig `json:"redis"`
Kafka KafkaConfig `json:"kafka"`
Auth AuthConfig `json:"auth"`
@@ -155,6 +156,16 @@ func applyEnv(cfg *Config) {
setString(&cfg.MySQL.DSN, "IM_WORKBENCH_MYSQL_DSN")
setInt(&cfg.MySQL.MaxOpenConns, "IM_WORKBENCH_MYSQL_MAX_OPEN_CONNS")
setInt(&cfg.MySQL.MaxIdleConns, "IM_WORKBENCH_MYSQL_MAX_IDLE_CONNS")
setString(&cfg.CoreMySQL.DSN, "IM_WORKBENCH_CORE_MYSQL_DSN")
if strings.TrimSpace(cfg.CoreMySQL.DSN) == "" {
cfg.CoreMySQL.DSN = cfg.MySQL.DSN
}
if cfg.CoreMySQL.MaxOpenConns == 0 {
cfg.CoreMySQL.MaxOpenConns = cfg.MySQL.MaxOpenConns
}
if cfg.CoreMySQL.MaxIdleConns == 0 {
cfg.CoreMySQL.MaxIdleConns = cfg.MySQL.MaxIdleConns
}
setString(&cfg.Redis.Addr, "IM_WORKBENCH_REDIS_ADDR")
setString(&cfg.Redis.Username, "IM_WORKBENCH_REDIS_USERNAME")
setString(&cfg.Redis.Password, "IM_WORKBENCH_REDIS_PASSWORD")
@@ -240,6 +251,10 @@ func (c *Config) validate() error {
if strings.EqualFold(strings.TrimSpace(parsedDSN.DBName), "tinode") {
return errors.New("mysql database tinode is deprecated; use workbench")
}
coreMySQLDSN := strings.TrimPrefix(strings.TrimSpace(c.CoreMySQL.DSN), "mysql://")
if _, err := mysqlDriver.ParseDSN(coreMySQLDSN); err != nil {
return fmt.Errorf("core mysql dsn is invalid: %w", err)
}
if strings.TrimSpace(c.Redis.Addr) == "" && !c.Auth.SkipSessionInDev {
return errors.New("redis addr is required")
}
internal/config/config_test.go
+59
View File
@@ -1,6 +1,8 @@
package config
import (
"os"
"path/filepath"
"strings"
"testing"
)
@@ -39,3 +41,60 @@ func TestApplyEnvOverridesMySQLPool(t *testing.T) {
t.Fatalf("mysql pool = %d/%d, want 12/4", cfg.MySQL.MaxOpenConns, cfg.MySQL.MaxIdleConns)
}
}
func TestLoadFallsBackCoreMySQLToPrimaryDSN(t *testing.T) {
path := filepath.Join(t.TempDir(), "workbench.json")
if err := os.WriteFile(path, []byte(`{"mysql":{"dsn":"root:root@tcp(127.0.0.1:3306)/workbench?parseTime=true"}}`), 0o600); err != nil {
t.Fatal(err)
}
t.Setenv("IM_WORKBENCH_CONFIG", path)
t.Setenv("IM_WORKBENCH_MYSQL_DSN", "")
t.Setenv("IM_WORKBENCH_CORE_MYSQL_DSN", "")
cfg, err := Load()
if err != nil {
t.Fatal(err)
}
if cfg.CoreMySQL.DSN != cfg.MySQL.DSN {
t.Fatalf("core_mysql.dsn = %q, want primary mysql.dsn %q", cfg.CoreMySQL.DSN, cfg.MySQL.DSN)
}
}
func TestLoadUsesCoreMySQLDSNFromEnvironment(t *testing.T) {
path := filepath.Join(t.TempDir(), "workbench.json")
if err := os.WriteFile(path, []byte(`{"mysql":{"dsn":"root:root@tcp(127.0.0.1:3306)/workbench?parseTime=true"}}`), 0o600); err != nil {
t.Fatal(err)
}
t.Setenv("IM_WORKBENCH_CONFIG", path)
t.Setenv("IM_WORKBENCH_MYSQL_DSN", "")
t.Setenv("IM_WORKBENCH_CORE_MYSQL_DSN", "root:root@tcp(127.0.0.1:3306)/meeting_im?parseTime=true")
cfg, err := Load()
if err != nil {
t.Fatal(err)
}
if !strings.Contains(cfg.CoreMySQL.DSN, "/meeting_im?") {
t.Fatalf("core_mysql.dsn = %q, want meeting_im", cfg.CoreMySQL.DSN)
}
}
func TestLoadPreservesCoreMySQLDSNFromJSON(t *testing.T) {
path := filepath.Join(t.TempDir(), "workbench.json")
if err := os.WriteFile(path, []byte(`{"mysql":{"dsn":"root:root@tcp(127.0.0.1:3306)/workbench?parseTime=true"},"core_mysql":{"dsn":"root:root@tcp(127.0.0.1:3306)/meeting_im?parseTime=true"}}`), 0o600); err != nil {
t.Fatal(err)
}
t.Setenv("IM_WORKBENCH_CONFIG", path)
t.Setenv("IM_WORKBENCH_MYSQL_DSN", "")
t.Setenv("IM_WORKBENCH_CORE_MYSQL_DSN", "")
cfg, err := Load()
if err != nil {
t.Fatal(err)
}
if !strings.Contains(cfg.MySQL.DSN, "/workbench?") {
t.Fatalf("mysql.dsn = %q, want workbench", cfg.MySQL.DSN)
}
if !strings.Contains(cfg.CoreMySQL.DSN, "/meeting_im?") {
t.Fatalf("core_mysql.dsn = %q, want meeting_im", cfg.CoreMySQL.DSN)
}
}
internal/store/core_mysql.go
+116
View File
@@ -0,0 +1,116 @@
package store
import (
"context"
"strings"
"git.118im.com/im_team/im_workbench/internal/config"
"github.com/jmoiron/sqlx"
)
// CoreStore provides read-only access to shared IM data.
type CoreStore struct {
db *sqlx.DB
}
func coreStoreOwnsDB(primaryDSN, coreDSN string) bool {
return strings.TrimSpace(primaryDSN) != strings.TrimSpace(coreDSN)
}
func openCoreDSN(cfg config.MySQLConfig) (*CoreStore, error) {
db, err := openMySQLDB(cfg)
if err != nil {
return nil, err
}
return &CoreStore{db: db}, nil
}
func (s *CoreStore) Close() error {
if s == nil || s.db == nil {
return nil
}
return s.db.Close()
}
func (s *CoreStore) fileUploadMetadata(ctx context.Context, tenantID int64, fileID string) fileUploadMetadata {
var meta fileUploadMetadata
decodedID, storageName := parseUploadFileID(fileID)
locationCandidates := uploadLocationCandidates(fileID, storageName)
if decodedID <= 0 && len(locationCandidates) == 0 {
return meta
}
if decodedID > 0 {
err := s.db.GetContext(ctx, &meta, `SELECT IFNULL(mimetype,'') AS mimetype,IFNULL(size,0) AS size,IFNULL(etag,'') AS etag,IFNULL(location,'') AS location,IFNULL(original_name,'') AS original_name FROM im_fileuploads WHERE tenant_id=? AND id=? LIMIT 1`, tenantID, decodedID)
if err == nil {
return meta
}
}
for _, candidate := range locationCandidates {
err := s.db.GetContext(ctx, &meta, `SELECT IFNULL(mimetype,'') AS mimetype,IFNULL(size,0) AS size,IFNULL(etag,'') AS etag,IFNULL(location,'') AS location,IFNULL(original_name,'') AS original_name
FROM im_fileuploads WHERE tenant_id=? AND (location=? OR location LIKE ?) ORDER BY createdat DESC LIMIT 1`, tenantID, candidate, "%/"+candidate)
if err == nil {
return meta
}
}
return meta
}
func (s *CoreStore) populateUserSummaries(ctx context.Context, tenantID int64, ids []int64, result map[int64]UserSummary) error {
query, args, err := sqlx.In(`SELECT user_id,nickname AS display_name,IFNULL(avatar_file_id,'') AS avatar_file_id
FROM im_user_profiles WHERE tenant_id=? AND user_id IN (?) ORDER BY user_id ASC`, tenantID, ids)
if err != nil {
return err
}
var profileRows []struct {
UserID int64 `db:"user_id"`
DisplayName string `db:"display_name"`
AvatarFileID string `db:"avatar_file_id"`
}
if err := s.db.SelectContext(ctx, &profileRows, s.db.Rebind(query), args...); err != nil && !isMissingTable(err) {
return err
}
for _, row := range profileRows {
summary := result[row.UserID]
if summary.UserID == 0 {
summary.UserID = row.UserID
}
if summary.DisplayName == "" {
summary.DisplayName = strings.TrimSpace(row.DisplayName)
}
if summary.AvatarFileID == "" {
summary.AvatarFileID = row.AvatarFileID
}
result[row.UserID] = summary
}
query, args, err = sqlx.In(`SELECT id AS user_id,
COALESCE(
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.display_name')),''),
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.fn')),''),
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.nickname')),''),
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.name')),''),
''
) AS display_name
FROM im_users WHERE tenant_id=? AND id IN (?)`, tenantID, ids)
if err != nil {
return err
}
var userRows []struct {
UserID int64 `db:"user_id"`
DisplayName string `db:"display_name"`
}
if err := s.db.SelectContext(ctx, &userRows, s.db.Rebind(query), args...); err != nil && !isMissingTable(err) {
return err
}
for _, row := range userRows {
summary := result[row.UserID]
if summary.UserID == 0 {
summary.UserID = row.UserID
}
if summary.DisplayName == "" {
summary.DisplayName = strings.TrimSpace(row.DisplayName)
}
result[row.UserID] = summary
}
return nil
}
internal/store/core_mysql_test.go
+23
View File
@@ -0,0 +1,23 @@
package store
import "testing"
func TestCoreStoreOwnsDBOnlyForDifferentDSN(t *testing.T) {
tests := []struct {
name string
primaryDSN string
coreDSN string
wantOwnDB bool
}{
{name: "same", primaryDSN: "root@tcp(db:3306)/meeting_im", coreDSN: "root@tcp(db:3306)/meeting_im", wantOwnDB: false},
{name: "same with whitespace", primaryDSN: " root@tcp(db:3306)/meeting_im ", coreDSN: "root@tcp(db:3306)/meeting_im", wantOwnDB: false},
{name: "different", primaryDSN: "root@tcp(db:3306)/workbench", coreDSN: "root@tcp(db:3306)/meeting_im", wantOwnDB: true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := coreStoreOwnsDB(tt.primaryDSN, tt.coreDSN); got != tt.wantOwnDB {
t.Fatalf("coreStoreOwnsDB() = %t, want %t", got, tt.wantOwnDB)
}
})
}
}
internal/store/mysql.go
+31 -76
View File
@@ -24,6 +24,7 @@ import (
type Store struct {
db *sqlx.DB
dsn string
core *CoreStore
tenantID int64
tenantStores map[int64]*Store
tenantStoreMu sync.Mutex
@@ -57,16 +58,34 @@ func SetUIDKey(encoded string) error {
return nil
}
func Open(cfg config.MySQLConfig) (*Store, error) {
func Open(cfg, coreCfg config.MySQLConfig) (*Store, error) {
st, err := openDSN(cfg)
if err != nil {
return nil, err
}
if coreStoreOwnsDB(cfg.DSN, coreCfg.DSN) {
core, err := openCoreDSN(coreCfg)
if err != nil {
_ = st.Close()
return nil, err
}
st.core = core
} else {
st.core = &CoreStore{db: st.db}
}
st.tenantStores = make(map[int64]*Store)
return st, nil
}
func openDSN(cfg config.MySQLConfig) (*Store, error) {
db, err := openMySQLDB(cfg)
if err != nil {
return nil, err
}
return &Store{db: db, dsn: cfg.DSN}, nil
}
func openMySQLDB(cfg config.MySQLConfig) (*sqlx.DB, error) {
db, err := sqlx.Open("mysql", cfg.DSN)
if err != nil {
return nil, err
@@ -80,7 +99,7 @@ func openDSN(cfg config.MySQLConfig) (*Store, error) {
_ = db.Close()
return nil, err
}
return &Store{db: db, dsn: cfg.DSN}, nil
return db, nil
}
func (s *Store) tenant() int64 {
@@ -102,6 +121,9 @@ func (s *Store) Close() error {
delete(s.tenantStores, id)
}
s.tenantStoreMu.Unlock()
if s.core != nil && s.core.db != s.db {
_ = s.core.Close()
}
return s.db.Close()
}
@@ -124,7 +146,7 @@ func (s *Store) ForTenant(ctx context.Context, tenantID int64, tenantCode string
s.tenantStoreMu.Unlock()
return st, nil
}
st := &Store{db: s.db, dsn: s.dsn, tenantID: cacheID}
st := &Store{db: s.db, dsn: s.dsn, core: s.core, tenantID: cacheID}
s.tenantStores[cacheID] = st
s.tenantStoreMu.Unlock()
return st, nil
@@ -648,26 +670,10 @@ type fileUploadMetadata struct {
}
func (s *Store) fileUploadMetadata(ctx context.Context, fileID string) fileUploadMetadata {
var meta fileUploadMetadata
decodedID, storageName := parseUploadFileID(fileID)
locationCandidates := uploadLocationCandidates(fileID, storageName)
if decodedID <= 0 && len(locationCandidates) == 0 {
return meta
if s.core == nil {
return fileUploadMetadata{}
}
if decodedID > 0 {
err := s.db.GetContext(ctx, &meta, `SELECT IFNULL(mimetype,'') AS mimetype,IFNULL(size,0) AS size,IFNULL(etag,'') AS etag,IFNULL(location,'') AS location,IFNULL(original_name,'') AS original_name FROM im_fileuploads WHERE tenant_id=? AND id=? LIMIT 1`, s.tenant(), decodedID)
if err == nil {
return meta
}
}
for _, candidate := range locationCandidates {
err := s.db.GetContext(ctx, &meta, `SELECT IFNULL(mimetype,'') AS mimetype,IFNULL(size,0) AS size,IFNULL(etag,'') AS etag,IFNULL(location,'') AS location,IFNULL(original_name,'') AS original_name
FROM im_fileuploads WHERE tenant_id=? AND (location=? OR location LIKE ?) ORDER BY createdat DESC LIMIT 1`, s.tenant(), candidate, "%/"+candidate)
if err == nil {
return meta
}
}
return meta
return s.core.fileUploadMetadata(ctx, s.tenant(), fileID)
}
func parseUploadFileID(fileID string) (int64, string) {
@@ -1206,61 +1212,10 @@ func (s *Store) UserSummaries(ctx context.Context, userIDs []int64) (map[int64]U
result[row.UserID] = UserSummary{UserID: row.UserID, DisplayName: strings.TrimSpace(row.DisplayName), AvatarFileID: row.AvatarFileID}
}
query, args, err = sqlx.In(`SELECT user_id,nickname AS display_name,IFNULL(avatar_file_id,'') AS avatar_file_id
FROM im_user_profiles WHERE tenant_id=? AND user_id IN (?) ORDER BY user_id ASC`, s.tenant(), ids)
if err != nil {
return nil, err
}
var profileRows []struct {
UserID int64 `db:"user_id"`
DisplayName string `db:"display_name"`
AvatarFileID string `db:"avatar_file_id"`
}
if err := s.db.SelectContext(ctx, &profileRows, s.db.Rebind(query), args...); err != nil && !isMissingTable(err) {
return nil, err
}
for _, row := range profileRows {
summary := result[row.UserID]
if summary.UserID == 0 {
summary.UserID = row.UserID
if s.core != nil {
if err := s.core.populateUserSummaries(ctx, s.tenant(), ids, result); err != nil {
return nil, err
}
if summary.DisplayName == "" {
summary.DisplayName = strings.TrimSpace(row.DisplayName)
}
if summary.AvatarFileID == "" {
summary.AvatarFileID = row.AvatarFileID
}
result[row.UserID] = summary
}
query, args, err = sqlx.In(`SELECT id AS user_id,
COALESCE(
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.display_name')),''),
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.fn')),''),
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.nickname')),''),
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.name')),''),
''
) AS display_name
FROM im_users WHERE tenant_id=? AND id IN (?)`, s.tenant(), ids)
if err != nil {
return nil, err
}
var userRows []struct {
UserID int64 `db:"user_id"`
DisplayName string `db:"display_name"`
}
if err := s.db.SelectContext(ctx, &userRows, s.db.Rebind(query), args...); err != nil && !isMissingTable(err) {
return nil, err
}
for _, row := range userRows {
summary := result[row.UserID]
if summary.UserID == 0 {
summary.UserID = row.UserID
}
if summary.DisplayName == "" {
summary.DisplayName = strings.TrimSpace(row.DisplayName)
}
result[row.UserID] = summary
}
for _, id := range ids {