From 8d32376de8dcc8953937664b835f4ecc137103a3 Mon Sep 17 00:00:00 2001 From: Your Name Date: Tue, 6 Oct 2026 23:40:01 +0800 Subject: [PATCH] feat: split workbench core mysql access --- internal/app/app.go | 2 +- internal/config/config.go | 15 ++++ internal/config/config_test.go | 59 +++++++++++++++ internal/store/core_mysql.go | 116 ++++++++++++++++++++++++++++++ internal/store/core_mysql_test.go | 23 ++++++ internal/store/mysql.go | 107 ++++++++------------------- 6 files changed, 245 insertions(+), 77 deletions(-) create mode 100644 internal/store/core_mysql.go create mode 100644 internal/store/core_mysql_test.go diff --git a/internal/app/app.go b/internal/app/app.go index 7e2b834..89850f2 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -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 } diff --git a/internal/config/config.go b/internal/config/config.go index b9ce6a8..2642dba 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -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") } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 7aced48..4cbb5ba 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -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) + } +} diff --git a/internal/store/core_mysql.go b/internal/store/core_mysql.go new file mode 100644 index 0000000..69df8e9 --- /dev/null +++ b/internal/store/core_mysql.go @@ -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 +} diff --git a/internal/store/core_mysql_test.go b/internal/store/core_mysql_test.go new file mode 100644 index 0000000..ea0f7fe --- /dev/null +++ b/internal/store/core_mysql_test.go @@ -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) + } + }) + } +} diff --git a/internal/store/mysql.go b/internal/store/mysql.go index 914a62e..e324ba8 100644 --- a/internal/store/mysql.go +++ b/internal/store/mysql.go @@ -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 {