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

feat: add im tenant db routing and chapter seven flows

Browse Source
This commit is contained in:
richjohnrambo
2026-08-13 11:30:48 +08:00
parent 774a448c12
commit 31f82d9c41
19 changed files with 5261 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/adapter.go
+261 -24
View File
@@ -13,11 +13,13 @@ import (
"sort"
"strconv"
"strings"
"sync"
"time"
ms "github.com/go-sql-driver/mysql"
"github.com/jmoiron/sqlx"
"github.com/tinode/chat/server/auth"
dbadapter "github.com/tinode/chat/server/db"
"github.com/tinode/chat/server/db/common"
"github.com/tinode/chat/server/store"
t "github.com/tinode/chat/server/store/types"
@@ -25,9 +27,10 @@ import (
// adapter holds MySQL connection data.
type adapter struct {
db *sqlx.DB
dsn string
dbName string
db *sqlx.DB
registryDB *sqlx.DB
dsn string
dbName string
// Maximum number of records to return
maxResults int
// Maximum number of message records to return
@@ -38,6 +41,12 @@ type adapter struct {
sqlTimeout time.Duration
// DB transaction timeout.
txTimeout time.Duration
baseConfig *configType
tenantDBRouting bool
tenantRegistryDB string
tenantAdapters map[t.TenantID]*adapter
tenantMu sync.Mutex
}
const (
@@ -78,6 +87,11 @@ type configType struct {
// DB request timeout (in seconds).
// If 0 (or negative), no timeout is applied.
SqlTimeout int `json:"sql_timeout,omitempty"`
// TenantRegistryDB is the public backend database, e.g. hotgo.
TenantRegistryDB string `json:"tenant_registry_db,omitempty"`
// TenantDBRouting enables "one tenant, one database" routing.
TenantDBRouting bool `json:"tenant_db_routing,omitempty"`
}
func (a *adapter) getContext() (context.Context, context.CancelFunc) {
@@ -94,6 +108,44 @@ func (a *adapter) getContextForTx() (context.Context, context.CancelFunc) {
return context.Background(), nil
}
func openMySQLDB(config configType, dsn, dbName string) (*sqlx.DB, string, error) {
if strings.TrimSpace(dsn) != "" {
cfg, err := ms.ParseDSN(strings.TrimPrefix(dsn, "mysql://"))
if err != nil {
return nil, "", err
}
if dbName != "" {
cfg.DBName = dbName
}
dsn = cfg.FormatDSN()
dbName = cfg.DBName
} else {
cfg := config.Config
if dbName != "" {
cfg.DBName = dbName
}
if cfg.DBName == "" {
cfg.DBName = defaultDatabase
}
dsn = cfg.FormatDSN()
dbName = cfg.DBName
}
db, err := sqlx.Open("mysql", dsn)
if err != nil {
return nil, "", err
}
if config.MaxOpenConns > 0 {
db.SetMaxOpenConns(config.MaxOpenConns)
}
if config.MaxIdleConns > 0 {
db.SetMaxIdleConns(config.MaxIdleConns)
}
if config.ConnMaxLifetime > 0 {
db.SetConnMaxLifetime(time.Duration(config.ConnMaxLifetime) * time.Second)
}
return db, dbName, nil
}
// Open initializes database session
func (a *adapter) Open(jsonconfig json.RawMessage) error {
if a.db != nil {
@@ -139,6 +191,11 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error {
if a.dbName == "" {
a.dbName = defaultDatabase
}
config.Config.DBName = a.dbName
a.baseConfig = &config
a.tenantDBRouting = config.TenantDBRouting
a.tenantRegistryDB = strings.TrimSpace(config.TenantRegistryDB)
a.tenantAdapters = make(map[t.TenantID]*adapter)
if a.maxResults <= 0 {
a.maxResults = defaultMaxResults
@@ -149,7 +206,7 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error {
}
// This just initializes the driver but does not open the network connection.
a.db, err = sqlx.Open("mysql", a.dsn)
a.db, a.dbName, err = openMySQLDB(config, a.dsn, a.dbName)
if err != nil {
return err
}
@@ -162,20 +219,20 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error {
err = nil
}
if err == nil {
if config.MaxOpenConns > 0 {
a.db.SetMaxOpenConns(config.MaxOpenConns)
}
if config.MaxIdleConns > 0 {
a.db.SetMaxIdleConns(config.MaxIdleConns)
}
if config.ConnMaxLifetime > 0 {
a.db.SetConnMaxLifetime(time.Duration(config.ConnMaxLifetime) * time.Second)
}
if config.SqlTimeout > 0 {
a.sqlTimeout = time.Duration(config.SqlTimeout) * time.Second
// We allocate txTimeoutMultiplier times sqlTimeout for transactions.
a.txTimeout = time.Duration(float64(config.SqlTimeout)*txTimeoutMultiplier) * time.Second
}
if a.tenantRegistryDB != "" && a.tenantRegistryDB != a.dbName {
a.registryDB, _, err = openMySQLDB(config, a.dsn, a.tenantRegistryDB)
if err == nil {
err = a.registryDB.Ping()
}
if err != nil {
return errors.New("mysql adapter failed to open tenant registry database: " + err.Error())
}
}
}
return err
}
@@ -183,8 +240,26 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error {
// Close closes the underlying database connection
func (a *adapter) Close() error {
var err error
a.tenantMu.Lock()
for tenantID, tenantAdp := range a.tenantAdapters {
if tenantAdp != nil && tenantAdp.db != nil {
if closeErr := tenantAdp.db.Close(); err == nil {
err = closeErr
}
}
delete(a.tenantAdapters, tenantID)
}
a.tenantMu.Unlock()
if a.registryDB != nil {
if closeErr := a.registryDB.Close(); err == nil {
err = closeErr
}
a.registryDB = nil
}
if a.db != nil {
err = a.db.Close()
if closeErr := a.db.Close(); err == nil {
err = closeErr
}
a.db = nil
a.version = -1
}
@@ -285,18 +360,112 @@ func (a *adapter) TenantGetByCode(code string) (*t.Tenant, error) {
}
var tenant t.Tenant
err := a.db.GetContext(ctx, &tenant,
`SELECT id, code, name, tenant_desc, state, created_at, created_by, updated_at, updated_by
db := a.db
if a.registryDB != nil {
db = a.registryDB
}
err := db.GetContext(ctx, &tenant,
`SELECT id, code, name, tenant_desc, IFNULL(db_name,'') AS db_name, IFNULL(db_dsn,'') AS db_dsn,
state, created_at, created_by, updated_at, updated_by
FROM im_tenant WHERE code=? LIMIT 1`, code)
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}
if isUnknownColumn(err) {
err = db.GetContext(ctx, &tenant,
`SELECT id, code, name, tenant_desc, state, created_at, created_by, updated_at, updated_by
FROM im_tenant WHERE code=? LIMIT 1`, code)
}
if err != nil {
return nil, err
}
return &tenant, nil
}
func (a *adapter) TenantAdapterFor(tenantID t.TenantID) (dbadapter.TenantBusinessAdapter, dbadapter.TenantAuthAdapter, error) {
if !a.tenantDBRouting || tenantID == defaultTenantID {
return a, a, nil
}
a.tenantMu.Lock()
if tenantAdp := a.tenantAdapters[tenantID]; tenantAdp != nil {
a.tenantMu.Unlock()
return tenantAdp, tenantAdp, nil
}
a.tenantMu.Unlock()
tenant, err := a.tenantByID(tenantID)
if err != nil {
return nil, nil, err
}
dbName := strings.TrimSpace(tenant.DBName)
dsn := strings.TrimSpace(tenant.DBDSN)
if dbName == "" && dsn == "" {
return a, a, nil
}
if a.baseConfig == nil {
return nil, nil, t.ErrFailed
}
db, resolvedDBName, err := openMySQLDB(*a.baseConfig, mysqlDefaultString(dsn, a.dsn), dbName)
if err != nil {
return nil, nil, err
}
if err = db.Ping(); err != nil {
db.Close()
return nil, nil, err
}
tenantAdp := &adapter{
db: db,
dsn: mysqlDefaultString(dsn, a.dsn),
dbName: resolvedDBName,
maxResults: a.maxResults,
maxMessageResults: a.maxMessageResults,
version: a.version,
sqlTimeout: a.sqlTimeout,
txTimeout: a.txTimeout,
baseConfig: a.baseConfig,
}
a.tenantMu.Lock()
if existing := a.tenantAdapters[tenantID]; existing != nil {
a.tenantMu.Unlock()
db.Close()
return existing, existing, nil
}
a.tenantAdapters[tenantID] = tenantAdp
a.tenantMu.Unlock()
return tenantAdp, tenantAdp, nil
}
func (a *adapter) tenantByID(tenantID t.TenantID) (*t.Tenant, error) {
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
db := a.db
if a.registryDB != nil {
db = a.registryDB
}
var tenant t.Tenant
err := db.GetContext(ctx, &tenant,
`SELECT id, code, name, tenant_desc, IFNULL(db_name,'') AS db_name, IFNULL(db_dsn,'') AS db_dsn,
state, created_at, created_by, updated_at, updated_by
FROM im_tenant WHERE id=? LIMIT 1`, tenantID)
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}
if isUnknownColumn(err) {
err = db.GetContext(ctx, &tenant,
`SELECT id, code, name, tenant_desc, state, created_at, created_by, updated_at, updated_by
FROM im_tenant WHERE id=? LIMIT 1`, tenantID)
}
if err != nil {
return nil, err
}
if !tenant.IsActive() {
return nil, t.ErrPermissionDenied
}
return &tenant, nil
}
// CreateDb initializes the storage.
func (a *adapter) CreateDb(reset bool) error {
var err error
@@ -348,6 +517,8 @@ func (a *adapter) CreateDb(reset bool) error {
code VARCHAR(64) NOT NULL,
name VARCHAR(128) NOT NULL,
tenant_desc VARCHAR(256),
db_name VARCHAR(128) NOT NULL DEFAULT '',
db_dsn VARCHAR(1024) NOT NULL DEFAULT '',
state SMALLINT NOT NULL,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
created_by BIGINT NOT NULL,
@@ -407,8 +578,8 @@ func (a *adapter) CreateDb(reset bool) error {
}
if _, err = tx.Exec(
`INSERT INTO im_tenant(code,name,tenant_desc,state,created_by,updated_by)
VALUES('default','Default Tenant',NULL,?,0,0)`, t.TenantStateActive); err != nil {
`INSERT INTO im_tenant(code,name,tenant_desc,db_name,db_dsn,state,created_by,updated_by)
VALUES('default','Default Tenant',NULL,'tinode','',?,0,0)`, t.TenantStateActive); err != nil {
return err
}
@@ -997,10 +1168,12 @@ func (a *adapter) UpgradeDb() error {
if _, err := a.db.Exec(
`CREATE TABLE im_tenant(
id BIGINT NOT NULL AUTO_INCREMENT,
code VARCHAR(64) NOT NULL,
name VARCHAR(128) NOT NULL,
tenant_desc VARCHAR(256),
state SMALLINT NOT NULL,
code VARCHAR(64) NOT NULL,
name VARCHAR(128) NOT NULL,
tenant_desc VARCHAR(256),
db_name VARCHAR(128) NOT NULL DEFAULT '',
db_dsn VARCHAR(1024) NOT NULL DEFAULT '',
state SMALLINT NOT NULL,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
created_by BIGINT NOT NULL,
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
@@ -1030,8 +1203,8 @@ func (a *adapter) UpgradeDb() error {
}
if _, err := a.db.Exec(
`INSERT INTO im_tenant(code,name,tenant_desc,state,created_by,updated_by)
VALUES('default','Default Tenant',NULL,?,0,0)`, t.TenantStateActive); err != nil {
`INSERT INTO im_tenant(code,name,tenant_desc,db_name,db_dsn,state,created_by,updated_by)
VALUES('default','Default Tenant',NULL,'tinode','',?,0,0)`, t.TenantStateActive); err != nil {
return err
}
@@ -1162,6 +1335,10 @@ func (a *adapter) UpgradeDb() error {
}
}
if err := ensureTenantRoutingColumns(a); err != nil {
return err
}
if a.version != adpVersion {
return errors.New("Failed to perform database upgrade to version " + strconv.Itoa(adpVersion) +
". DB is still at " + strconv.Itoa(a.version))
@@ -1191,6 +1368,20 @@ func createIMNotificationTables(tx *sql.Tx) error {
return err
}
func ensureTenantRoutingColumns(a *adapter) error {
statements := []string{
"ALTER TABLE im_tenant ADD COLUMN db_name VARCHAR(128) NOT NULL DEFAULT '' AFTER tenant_desc",
"ALTER TABLE im_tenant ADD COLUMN db_dsn VARCHAR(1024) NOT NULL DEFAULT '' AFTER db_name",
"UPDATE im_tenant SET db_name='tinode' WHERE code='default' AND db_name=''",
}
for _, stmt := range statements {
if _, err := a.db.Exec(stmt); err != nil && !isDuplicateColumn(err) {
return err
}
}
return nil
}
func upgradeIMMessageActions(a *adapter) error {
statements := []string{
"ALTER TABLE messages ADD COLUMN version INT NOT NULL DEFAULT 1 AFTER client_msg_id",
@@ -1272,6 +1463,29 @@ func createIMRelationTables(exec sqlExecutor) error {
INDEX idx_im_user_blocks_blocked(tenant_id, blocked_id),
CONSTRAINT fk_im_user_blocks_tenant FOREIGN KEY(tenant_id) REFERENCES im_tenant(id) ON DELETE CASCADE
)`,
`CREATE TABLE im_favorites(
id BIGINT NOT NULL AUTO_INCREMENT,
tenant_id BIGINT NOT NULL,
favorite_id VARCHAR(64) NOT NULL,
user_id BIGINT NOT NULL,
item_type VARCHAR(16) NOT NULL,
source_topic CHAR(25) NOT NULL,
source_seq INT NOT NULL,
snapshot JSON,
attachment_refs JSON,
availability VARCHAR(16) NOT NULL DEFAULT 'available',
idempotency_key VARCHAR(128) NOT NULL DEFAULT '',
collected_at DATETIME(3) NOT NULL,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id),
UNIQUE INDEX uk_im_favorites_favorite(tenant_id, favorite_id),
UNIQUE INDEX uk_im_favorites_source(tenant_id, user_id, source_topic, source_seq),
UNIQUE INDEX uk_im_favorites_idempotency(tenant_id, user_id, idempotency_key),
INDEX idx_im_favorites_inbox(tenant_id, user_id, id DESC),
INDEX idx_im_favorites_type(tenant_id, user_id, item_type, id DESC),
CONSTRAINT fk_im_favorites_tenant FOREIGN KEY(tenant_id) REFERENCES im_tenant(id) ON DELETE CASCADE
)`,
}
for _, stmt := range statements {
if _, err := exec.Exec(stmt); err != nil {
@@ -4608,6 +4822,29 @@ func isMissingDb(err error) bool {
return ok && myerr.Number == 1049
}
func isUnknownColumn(err error) bool {
if err == nil {
return false
}
myerr, ok := err.(*ms.MySQLError)
return ok && myerr.Number == 1054
}
func isDuplicateColumn(err error) bool {
if err == nil {
return false
}
myerr, ok := err.(*ms.MySQLError)
return ok && myerr.Number == 1060
}
func mysqlDefaultString(value, fallback string) string {
if strings.TrimSpace(value) == "" {
return fallback
}
return strings.TrimSpace(value)
}
// GetTestAdapter returns an adapter object. Useful for running tests.
func GetTestAdapter() *adapter {
return &adapter{}