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

fix(mysql): adapt tenant shared database fields

Browse Source
This commit is contained in:
richjohnrambo
2026-09-22 14:40:15 +08:00
parent 865b436264
commit 2e0612acbb
11 changed files with 861 additions and 528 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/adapter.go
+268 -46
View File
@@ -45,6 +45,7 @@ type adapter struct {
baseConfig *configType
tenantDBRouting bool
tenantScopedDB bool
tenantID t.TenantID
tenantRegistryDB string
tenantAdapters map[t.TenantID]*adapter
tenantMu sync.Mutex
@@ -250,6 +251,7 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error {
config.Config.DBName = a.dbName
a.baseConfig = &config
a.tenantDBRouting = config.TenantDBRouting
a.tenantID = defaultTenantID
a.tenantRegistryDB = strings.TrimSpace(config.TenantRegistryDB)
a.tenantAdapters = make(map[t.TenantID]*adapter)
@@ -296,7 +298,7 @@ func (a *adapter) Close() error {
var err error
a.tenantMu.Lock()
for tenantID, tenantAdp := range a.tenantAdapters {
if tenantAdp != nil && tenantAdp.db != nil {
if tenantAdp != nil && tenantAdp.db != nil && tenantAdp.db != a.db {
if closeErr := tenantAdp.db.Close(); err == nil {
err = closeErr
}
@@ -471,8 +473,27 @@ func (a *adapter) TenantListActive(limit int) ([]t.Tenant, error) {
func (a *adapter) TenantAdapterFor(tenantID t.TenantID) (dbadapter.TenantBusinessAdapter, dbadapter.TenantAuthAdapter, error) {
if !a.tenantDBRouting {
if a.tenantID != defaultTenantID && a.tenantID != tenantID {
return nil, nil, t.ErrMalformed
}
if a.registryDB != nil {
tenant, err := a.tenantByID(tenantID)
if err != nil {
return nil, nil, err
}
if tenant.State != t.TenantStateActive {
return nil, nil, t.ErrPermissionDenied
}
}
return a, a, nil
}
tenant, err := a.tenantByID(tenantID)
if err != nil {
return nil, nil, err
}
if tenant.State != t.TenantStateActive {
return nil, nil, t.ErrPermissionDenied
}
a.tenantMu.Lock()
if tenantAdp := a.tenantAdapters[tenantID]; tenantAdp != nil {
a.tenantMu.Unlock()
@@ -480,36 +501,15 @@ func (a *adapter) TenantAdapterFor(tenantID t.TenantID) (dbadapter.TenantBusines
}
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 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
}
if err = ensureUserPrivacySchema(db); err != nil {
db.Close()
return nil, nil, errors.New("mysql adapter failed to ensure user privacy schema for tenant: " + err.Error())
}
registryDB := a.registryDB
if registryDB == nil {
registryDB = a.db
}
tenantAdp := &adapter{
db: db,
db: a.db,
registryDB: registryDB,
dsn: mysqlDefaultString(dsn, a.dsn),
dbName: resolvedDBName,
dsn: a.dsn,
dbName: a.dbName,
maxResults: a.maxResults,
maxMessageResults: a.maxMessageResults,
version: a.version,
@@ -517,12 +517,14 @@ func (a *adapter) TenantAdapterFor(tenantID t.TenantID) (dbadapter.TenantBusines
txTimeout: a.txTimeout,
baseConfig: a.baseConfig,
tenantDBRouting: false,
tenantScopedDB: true,
tenantID: tenantID,
// A dedicated database is still tenant-owned. Keep tenant predicates and
// explicit tenant_id writes enabled even though databases are not shared.
tenantScopedDB: false,
}
a.tenantMu.Lock()
if existing := a.tenantAdapters[tenantID]; existing != nil {
a.tenantMu.Unlock()
db.Close()
return existing, existing, nil
}
a.tenantAdapters[tenantID] = tenantAdp
@@ -570,7 +572,7 @@ func createTenantRegistry(tx *sql.Tx, registryDB, businessDB string) error {
}
if _, err := tx.Exec(
`CREATE TABLE IF NOT EXISTS im_tenant(
id BIGINT NOT NULL AUTO_INCREMENT,
id BIGINT NOT NULL,
code VARCHAR(64) NOT NULL,
name VARCHAR(128) NOT NULL,
tenant_desc VARCHAR(256),
@@ -1429,18 +1431,35 @@ func (a *adapter) TenantAuthAddRecord(tenantID t.TenantID, uid t.Uid,
if tenantAdp != a {
return tenantAdp.TenantAuthAddRecord(tenantID, uid, authLvl, secret, expires)
}
var exp *time.Time
if !expires.IsZero() {
exp = &expires
}
var username string
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
if err = a.db.GetContext(ctx, &username, "SELECT username FROM im_users WHERE id=?", store.DecodeUid(uid)); err != nil {
if err = a.db.GetContext(ctx, &username, "SELECT username FROM im_users WHERE tenant_id=? AND id=?", tenantID, store.DecodeUid(uid)); err != nil {
if err == sql.ErrNoRows {
return t.ErrNotFound
}
return err
}
return a.AuthAddRecord(uid, "basic", username, authLvl, secret, expires)
authID, err := nextSnowflakeID()
if err != nil {
return err
}
_, err = a.db.ExecContext(ctx, "INSERT INTO im_auth(tenant_id,id,userid,authLvl,secret,expires) VALUES(?,?,?,?,?,?)",
tenantID, authID, store.DecodeUid(uid), authLvl, secret, exp)
if isIDOutOfRange(err) {
_, err = a.db.ExecContext(ctx, "INSERT INTO im_auth(tenant_id,userid,authLvl,secret,expires) VALUES(?,?,?,?,?)",
tenantID, store.DecodeUid(uid), authLvl, secret, exp)
}
if isDupe(err) {
return t.ErrDuplicate
}
return err
}
// TenantAuthDeleteRecord deletes the Basic authentication record.
@@ -1452,7 +1471,12 @@ func (a *adapter) TenantAuthDeleteRecord(tenantID t.TenantID, uid t.Uid) error {
if tenantAdp != a {
return tenantAdp.TenantAuthDeleteRecord(tenantID, uid)
}
return a.AuthDelScheme(uid, "basic")
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
_, err = a.db.ExecContext(ctx, "DELETE FROM im_auth WHERE tenant_id=? AND userid=?", tenantID, store.DecodeUid(uid))
return err
}
// TenantAuthUpdateRecord updates the Basic authentication record and optional username.
@@ -1465,7 +1489,43 @@ func (a *adapter) TenantAuthUpdateRecord(tenantID t.TenantID, uid t.Uid, usernam
if tenantAdp != a {
return tenantAdp.TenantAuthUpdateRecord(tenantID, uid, username, authLvl, secret, expires)
}
return a.AuthUpdRecord(uid, "basic", username, authLvl, secret, expires)
params := []string{"authLvl=?"}
args := []any{authLvl}
if len(secret) > 0 {
params = append(params, "secret=?")
args = append(args, secret)
}
if !expires.IsZero() {
params = append(params, "expires=?")
args = append(args, expires)
}
decodedUID := store.DecodeUid(uid)
ctx, cancel := a.getContextForTx()
if cancel != nil {
defer cancel()
}
tx, err := a.db.BeginTxx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
if username != "" {
if _, err = tx.ExecContext(ctx, "UPDATE im_users SET username=?,updatedat=? WHERE tenant_id=? AND id=?", username, t.TimeNow(), tenantID, decodedUID); err != nil {
if isDupe(err) {
return t.ErrDuplicate
}
return err
}
}
args = append(args, tenantID, decodedUID)
resp, err := tx.ExecContext(ctx, "UPDATE im_auth SET "+strings.Join(params, ",")+" WHERE tenant_id=? AND userid=?", args...)
if err != nil {
return err
}
if count, _ := resp.RowsAffected(); count <= 0 {
return t.ErrNotFound
}
return tx.Commit()
}
// TenantAuthGetRecord retrieves a user's Basic authentication record.
@@ -1479,7 +1539,29 @@ func (a *adapter) TenantAuthGetRecord(tenantID t.TenantID, uid t.Uid) (string,
if tenantAdp != a {
return tenantAdp.TenantAuthGetRecord(tenantID, uid)
}
return a.AuthGetRecord(uid, "basic")
var record struct {
Username string
Authlvl auth.Level
Secret []byte
Expires *time.Time
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
err = a.db.GetContext(ctx, &record, `SELECT u.username,a.secret,a.expires,a.authlvl
FROM im_auth AS a JOIN im_users AS u ON u.tenant_id=a.tenant_id AND u.id=a.userid
WHERE a.tenant_id=? AND a.userid=?`, tenantID, store.DecodeUid(uid))
if err == sql.ErrNoRows {
return "", auth.LevelNone, nil, expires, t.ErrNotFound
}
if err != nil {
return "", auth.LevelNone, nil, expires, err
}
if record.Expires != nil {
expires = *record.Expires
}
return record.Username, record.Authlvl, record.Secret, expires, nil
}
// TenantAuthGetByLogin retrieves a Basic authentication record by tenant-local login.
@@ -1493,7 +1575,34 @@ func (a *adapter) TenantAuthGetByLogin(tenantID t.TenantID, login string) (t.Uid
if tenantAdp != a {
return tenantAdp.TenantAuthGetByLogin(tenantID, login)
}
return a.AuthGetUniqueRecord(login)
var record struct {
Userid int64
Authlvl auth.Level
Secret []byte
Expires *time.Time
}
lookup := login
if len(login) == 11 && asciiDigits(login) {
lookup = "+86" + login
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
err = a.db.GetContext(ctx, &record, `SELECT a.userid,a.secret,a.expires,a.authlvl
FROM im_users AS u JOIN im_auth AS a ON a.tenant_id=u.tenant_id AND a.userid=u.id
WHERE u.tenant_id=? AND (u.username=? OR (u.phone IS NOT NULL AND u.phone=?))
ORDER BY (u.username=?) DESC LIMIT 1`, tenantID, login, lookup, login)
if err == sql.ErrNoRows {
return t.ZeroUid, auth.LevelNone, nil, expires, nil
}
if err != nil {
return t.ZeroUid, auth.LevelNone, nil, expires, err
}
if record.Expires != nil {
expires = *record.Expires
}
return store.EncodeUid(record.Userid), record.Authlvl, record.Secret, expires, nil
}
// Add user's authentication record
@@ -2011,9 +2120,9 @@ func (a *adapter) TenantUserGetByCred(tenantID t.TenantID, method, value string)
return t.ZeroUid, t.ErrMalformed
}
if tenantAdp != a {
return tenantAdp.UserGetByCred(method, value)
return tenantAdp.TenantUserGetByCred(tenantID, method, value)
}
if !a.tenantScopedDB && method == "tel" {
if method == "tel" {
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
@@ -2029,7 +2138,20 @@ func (a *adapter) TenantUserGetByCred(tenantID t.TenantID, method, value string)
}
return store.EncodeUid(decodedUID), nil
}
return a.UserGetByCred(method, value)
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
var decodedUID int64
err = a.db.GetContext(ctx, &decodedUID,
"SELECT userid FROM im_credentials WHERE tenant_id=? AND synthetic=?", tenantID, method+":"+value)
if err == sql.ErrNoRows {
return t.ZeroUid, nil
}
if err != nil {
return t.ZeroUid, err
}
return store.EncodeUid(decodedUID), nil
}
// UserUnreadCount returns the total number of unread messages in all topics with
@@ -3855,9 +3977,73 @@ func (a *adapter) TenantCredUpsert(tenantID t.TenantID, cred *t.Credential) (boo
return false, t.ErrMalformed
}
if tenantAdp != a {
return tenantAdp.CredUpsert(cred)
return tenantAdp.TenantCredUpsert(tenantID, cred)
}
return a.CredUpsert(cred)
if cred.Method == "tel" {
cred.Value, err = normalizeUserPhone(cred.Value)
if err != nil {
return false, err
}
}
ctx, cancel := a.getContextForTx()
if cancel != nil {
defer cancel()
}
tx, err := a.db.BeginTxx(ctx, nil)
if err != nil {
return false, err
}
defer tx.Rollback()
now := t.TimeNow()
userID := common.DecodeUidString(cred.User)
synth := cred.Method + ":" + cred.Value
if !cred.Done {
var done bool
err = tx.Get(&done, "SELECT done FROM im_credentials WHERE tenant_id=? AND synthetic=?", tenantID, synth)
if err == nil {
return false, t.ErrDuplicate
}
if err != sql.ErrNoRows {
return false, err
}
synth = cred.User + ":" + synth
if _, err = tx.Exec("UPDATE im_credentials SET deletedat=? WHERE tenant_id=? AND userid=? AND method=? AND done=FALSE",
now, tenantID, userID, cred.Method); err != nil {
return false, err
}
res, updateErr := tx.Exec("UPDATE im_credentials SET updatedat=?,deletedat=NULL,resp=?,done=FALSE WHERE tenant_id=? AND synthetic=?",
cred.UpdatedAt, cred.Resp, tenantID, synth)
if updateErr != nil {
return false, updateErr
}
if rows, _ := res.RowsAffected(); rows > 0 {
return false, tx.Commit()
}
} else if _, err = tx.Exec("DELETE FROM im_credentials WHERE tenant_id=? AND synthetic=?", tenantID, cred.User+":"+synth); err != nil {
return false, err
}
credID, err := nextSnowflakeID()
if err != nil {
return false, err
}
_, err = tx.Exec(`INSERT INTO im_credentials(tenant_id,id,createdat,updatedat,method,value,synthetic,userid,resp,done)
VALUES(?,?,?,?,?,?,?,?,?,?)`, tenantID, credID, cred.CreatedAt, cred.UpdatedAt, cred.Method, cred.Value, synth, userID, cred.Resp, cred.Done)
if isIDOutOfRange(err) {
_, err = tx.Exec(`INSERT INTO im_credentials(tenant_id,createdat,updatedat,method,value,synthetic,userid,resp,done)
VALUES(?,?,?,?,?,?,?,?,?)`, tenantID, cred.CreatedAt, cred.UpdatedAt, cred.Method, cred.Value, synth, userID, cred.Resp, cred.Done)
}
if err != nil {
if isDupe(err) {
return true, t.ErrDuplicate
}
return true, err
}
if cred.Done && cred.Method == "tel" {
if _, err = tx.Exec("UPDATE im_users SET phone=? WHERE tenant_id=? AND id=? AND (phone IS NULL OR phone='')", cred.Value, tenantID, userID); err != nil {
return true, err
}
}
return true, tx.Commit()
}
func tenantCredDel(tx *sqlx.Tx, tenantID t.TenantID, uid t.Uid, method, value string) error {
@@ -3989,7 +4175,7 @@ func (a *adapter) TenantCredDel(tenantID t.TenantID, uid t.Uid, method, value st
return t.ErrMalformed
}
if tenantAdp != a {
return tenantAdp.CredDel(uid, method, value)
return tenantAdp.TenantCredDel(tenantID, uid, method, value)
}
ctx, cancel := a.getContextForTx()
if cancel != nil {
@@ -4036,7 +4222,7 @@ func (a *adapter) TenantCredConfirm(tenantID t.TenantID, uid t.Uid, method strin
return t.ErrMalformed
}
if tenantAdp != a {
return tenantAdp.CredConfirm(uid, method)
return tenantAdp.TenantCredConfirm(tenantID, uid, method)
}
ctx, cancel := a.getContext()
if cancel != nil {
@@ -4076,7 +4262,7 @@ func (a *adapter) TenantCredFail(tenantID t.TenantID, uid t.Uid, method string)
return t.ErrMalformed
}
if tenantAdp != a {
return tenantAdp.CredFail(uid, method)
return tenantAdp.TenantCredFail(tenantID, uid, method)
}
ctx, cancel := a.getContext()
if cancel != nil {
@@ -4116,9 +4302,24 @@ func (a *adapter) TenantCredGetActive(tenantID t.TenantID, uid t.Uid, method str
return nil, t.ErrMalformed
}
if tenantAdp != a {
return tenantAdp.CredGetActive(uid, method)
return tenantAdp.TenantCredGetActive(tenantID, uid, method)
}
return a.CredGetActive(uid, method)
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
var cred t.Credential
err = a.db.GetContext(ctx, &cred, `SELECT createdat,updatedat,method,value,resp,done,retries
FROM im_credentials WHERE tenant_id=? AND userid=? AND deletedat IS NULL AND method=? AND done=FALSE`,
tenantID, store.DecodeUid(uid), method)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
cred.User = uid.String()
return &cred, nil
}
// CredGetAll returns credential records for the given user and method, all or validated only.
@@ -4159,9 +4360,30 @@ func (a *adapter) TenantCredGetAll(tenantID t.TenantID, uid t.Uid, method string
return nil, t.ErrMalformed
}
if tenantAdp != a {
return tenantAdp.CredGetAll(uid, method, validatedOnly)
return tenantAdp.TenantCredGetAll(tenantID, uid, method, validatedOnly)
}
return a.CredGetAll(uid, method, validatedOnly)
query := `SELECT createdat,updatedat,method,value,resp,done,retries
FROM im_credentials WHERE tenant_id=? AND userid=? AND deletedat IS NULL`
args := []any{tenantID, store.DecodeUid(uid)}
if method != "" {
query += " AND method=?"
args = append(args, method)
}
if validatedOnly {
query += " AND done=TRUE"
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
var credentials []t.Credential
if err = a.db.SelectContext(ctx, &credentials, query, args...); err != nil {
return nil, err
}
for i := range credentials {
credentials[i].User = uid.String()
}
return credentials, nil
}
// FileUploads
server/db/mysql/group_message_shards.go
+33 -47
View File
@@ -13,37 +13,35 @@ import (
t "github.com/tinode/chat/server/store/types"
)
const legacyGroupMessageTable = "im_group_messages"
func ensureGroupMessageShardInfrastructure(exec sqlExecutor) error {
func ensureGroupMessageShardInfrastructure(exec sqlExecutor, tenantID t.TenantID) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS im_group_message_shards (
shard_month CHAR(6) NOT NULL, table_name VARCHAR(64) NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, shard_month CHAR(6) NOT NULL, table_name VARCHAR(64) NOT NULL,
shard_start DATETIME(3) NOT NULL, shard_end DATETIME(3) NOT NULL,
status VARCHAR(16) NOT NULL DEFAULT 'prepared', row_count BIGINT NOT NULL DEFAULT 0,
min_created_at DATETIME(3), max_created_at DATETIME(3),
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(shard_month), UNIQUE INDEX uk_im_group_message_shards_table(table_name),
INDEX idx_im_group_message_shards_status(status, shard_month)
PRIMARY KEY(tenant_id, shard_month), UNIQUE INDEX uk_im_group_message_shards_table(tenant_id, table_name),
INDEX idx_im_group_message_shards_status(tenant_id, status, shard_month)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_message_topic_shards (
topic VARCHAR(64) NOT NULL, shard_month CHAR(6) NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, topic VARCHAR(64) NOT NULL, shard_month CHAR(6) NOT NULL,
min_seq BIGINT NOT NULL, max_seq BIGINT NOT NULL, message_count BIGINT NOT NULL DEFAULT 0,
min_created_at DATETIME(3) NOT NULL, max_created_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(topic, shard_month), INDEX idx_im_group_message_topic_shards_seq(topic, max_seq),
INDEX idx_im_group_message_topic_shards_month(shard_month, topic)
PRIMARY KEY(tenant_id, topic, shard_month), INDEX idx_im_group_message_topic_shards_seq(tenant_id, topic, max_seq),
INDEX idx_im_group_message_topic_shards_month(tenant_id, shard_month, topic)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_message_dedup (
topic VARCHAR(64) NOT NULL, from_user BIGINT NOT NULL, client_msg_id VARCHAR(64) NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, topic VARCHAR(64) NOT NULL, from_user BIGINT NOT NULL, client_msg_id VARCHAR(64) NOT NULL,
seqid BIGINT NOT NULL, shard_month CHAR(6) NOT NULL, message_id BIGINT,
status VARCHAR(16) NOT NULL DEFAULT 'pending', expires_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(topic, from_user, client_msg_id), INDEX idx_im_group_message_dedup_expires(expires_at),
INDEX idx_im_group_message_dedup_topic_seq(topic, seqid)
PRIMARY KEY(tenant_id, topic, from_user, client_msg_id), INDEX idx_im_group_message_dedup_expires(expires_at),
INDEX idx_im_group_message_dedup_topic_seq(tenant_id, topic, seqid)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
}
for _, statement := range statements {
@@ -52,10 +50,10 @@ func ensureGroupMessageShardInfrastructure(exec sqlExecutor) error {
}
}
return ensureGroupMessageShardWindow(exec, time.Now().UTC())
return ensureGroupMessageShardWindow(exec, tenantID, time.Now().UTC())
}
func ensureGroupMessageShardWindow(exec sqlExecutor, now time.Time) error {
func ensureGroupMessageShardWindow(exec sqlExecutor, tenantID t.TenantID, now time.Time) error {
first := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, time.UTC)
for _, offset := range []int{-1, 0, 1, 2} {
start := first.AddDate(0, offset, 0)
@@ -65,11 +63,11 @@ func ensureGroupMessageShardWindow(exec sqlExecutor, now time.Time) error {
} else if offset == 0 {
status = "writable"
}
if err := ensureGroupMessageShard(exec, start, status); err != nil {
if err := ensureGroupMessageShard(exec, tenantID, start, status); err != nil {
return err
}
}
_, err := exec.Exec("UPDATE im_group_message_shards SET status='sealed' WHERE status='sealing' AND shard_end<?", first.AddDate(0, 0, -7))
_, err := exec.Exec("UPDATE im_group_message_shards SET status='sealed' WHERE tenant_id=? AND status='sealing' AND shard_end<?", tenantID, first.AddDate(0, 0, -7))
return err
}
@@ -80,21 +78,21 @@ func (a *adapter) ensureCurrentGroupMessageShardWindow() error {
if a.groupShardWindow == month {
return nil
}
if err := ensureGroupMessageShardWindow(a.db, time.Now().UTC()); err != nil {
if err := ensureGroupMessageShardWindow(a.db, a.tenantID, time.Now().UTC()); err != nil {
return err
}
a.groupShardWindow = month
return nil
}
func ensureGroupMessageShard(exec sqlExecutor, start time.Time, status string) error {
func ensureGroupMessageShard(exec sqlExecutor, tenantID t.TenantID, start time.Time, status string) error {
month := start.UTC().Format("200601")
table, ok := groupMessageShardTable(month)
table, ok := tenantGroupMessagesTable(tenantID, month)
if !ok {
return fmt.Errorf("mysql: invalid group message shard month %q", month)
}
ddl := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (
message_id BIGINT NOT NULL, topic VARCHAR(64) NOT NULL, group_id BIGINT NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, message_id BIGINT NOT NULL, topic VARCHAR(64) NOT NULL, group_id BIGINT NOT NULL,
seqid BIGINT NOT NULL, from_user BIGINT NOT NULL, client_msg_id VARCHAR(64),
msg_type VARCHAR(32) NOT NULL DEFAULT 'text', status VARCHAR(32) NOT NULL DEFAULT 'normal',
content JSON, content_nonce VARBINARY(12), key_version INT NOT NULL DEFAULT 0,
@@ -117,28 +115,16 @@ func ensureGroupMessageShard(exec sqlExecutor, start time.Time, status string) e
}
end := start.AddDate(0, 1, 0)
_, err := exec.Exec(`INSERT INTO im_group_message_shards
(shard_month,table_name,shard_start,shard_end,status)
VALUES(?,?,?,?,?) ON DUPLICATE KEY UPDATE table_name=VALUES(table_name),
(tenant_id,shard_month,table_name,shard_start,shard_end,status)
VALUES(?,?,?,?,?,?) ON DUPLICATE KEY UPDATE tenant_id=VALUES(tenant_id),table_name=VALUES(table_name),
shard_start=VALUES(shard_start),shard_end=VALUES(shard_end),
status=IF(status IN ('sealed','archived'),status,VALUES(status))`,
month, table, start, end, status)
tenantID, month, table, start, end, status)
return err
}
func groupMessageShardTable(month string) (string, bool) {
if len(month) != 6 {
return "", false
}
for i := range month {
if month[i] < '0' || month[i] > '9' {
return "", false
}
}
parsed, err := time.Parse("200601", month)
if err != nil || parsed.Format("200601") != month {
return "", false
}
return "im_group_messages_" + month, true
func groupMessageShardTable(tenantID t.TenantID, month string) (string, bool) {
return tenantGroupMessagesTable(tenantID, month)
}
func groupMessageTablesForTopic(q sqlx.QueryerContext, ctx context.Context, tenantID t.TenantID, topic string) ([]string, error) {
@@ -151,7 +137,7 @@ func groupMessageTablesForTopic(q sqlx.QueryerContext, ctx context.Context, tena
routes = nil
if err = sqlx.SelectContext(ctx, q, &routes,
`SELECT shard_month AS month,min_seq,max_seq,message_count AS count
FROM im_group_message_topic_shards WHERE topic=? ORDER BY shard_month DESC`, topic); err != nil {
FROM im_group_message_topic_shards WHERE tenant_id=? AND topic=? ORDER BY shard_month DESC`, tenantID, topic); err != nil {
return nil, err
}
if redisAvailable {
@@ -162,11 +148,11 @@ func groupMessageTablesForTopic(q sqlx.QueryerContext, ctx context.Context, tena
}
tables := make([]string, 0, len(routes)+1)
for _, route := range routes {
if table, ok := groupMessageShardTable(route.Month); ok {
if table, ok := groupMessageShardTable(tenantID, route.Month); ok {
tables = append(tables, table)
}
}
return append(tables, legacyGroupMessageTable), nil
return tables, nil
}
func groupMessageTablesForSeq(q sqlx.QueryerContext, ctx context.Context, tenantID t.TenantID, topic string, seq int64) ([]string, error) {
@@ -179,8 +165,8 @@ func groupMessageTablesForSeq(q sqlx.QueryerContext, ctx context.Context, tenant
routes = nil
if err = sqlx.SelectContext(ctx, q, &routes,
`SELECT shard_month AS month,min_seq,max_seq,message_count AS count
FROM im_group_message_topic_shards WHERE topic=? AND min_seq<=? AND max_seq>=?
ORDER BY shard_month DESC`, topic, seq, seq); err != nil {
FROM im_group_message_topic_shards WHERE tenant_id=? AND topic=? AND min_seq<=? AND max_seq>=?
ORDER BY shard_month DESC`, tenantID, topic, seq, seq); err != nil {
return nil, err
}
if redisAvailable {
@@ -192,18 +178,18 @@ func groupMessageTablesForSeq(q sqlx.QueryerContext, ctx context.Context, tenant
tables := make([]string, 0, 2)
for _, route := range routes {
if route.MinSeq <= seq && route.MaxSeq >= seq {
if table, ok := groupMessageShardTable(route.Month); ok {
if table, ok := groupMessageShardTable(tenantID, route.Month); ok {
tables = append(tables, table)
}
}
}
return append(tables, legacyGroupMessageTable), nil
return tables, nil
}
func groupMessageMonthFromTable(table string) string {
const prefix = "im_group_messages_"
if len(table) == len(prefix)+6 && table[:len(prefix)] == prefix {
return table[len(prefix):]
const suffixLength = 6
if len(table) > suffixLength && validShardMonth(table[len(table)-suffixLength:]) {
return table[len(table)-suffixLength:]
}
return ""
}
server/db/mysql/group_schema.go
+15 -15
View File
@@ -4,12 +4,12 @@
package mysql
// ensureIMGroupBusinessTables installs the product group-chat schema in the
// currently selected tenant database. Tenant databases intentionally do not
// carry tenant_id columns.
// currently selected tenant database. Rows still carry tenant_id explicitly;
// database routing remains one tenant per database.
func ensureIMGroupBusinessTables(exec sqlExecutor) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS im_group_settings (
id BIGINT NOT NULL, topic VARCHAR(32) NOT NULL, owner_user_id BIGINT NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL, topic VARCHAR(32) NOT NULL, owner_user_id BIGINT NOT NULL,
name VARCHAR(128) NOT NULL DEFAULT '', avatar_file_id VARCHAR(64) NOT NULL DEFAULT '',
description VARCHAR(512) NOT NULL DEFAULT '', group_status VARCHAR(16) NOT NULL DEFAULT 'active',
join_approval TINYINT NOT NULL DEFAULT 0, mute_all TINYINT NOT NULL DEFAULT 0,
@@ -24,7 +24,7 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
INDEX idx_im_group_settings_status_updated(group_status, updated_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_messages (
id BIGINT NOT NULL AUTO_INCREMENT, topic VARCHAR(64) NOT NULL, group_id BIGINT NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, topic VARCHAR(64) NOT NULL, group_id BIGINT NOT NULL,
seqid BIGINT NOT NULL, from_user BIGINT NOT NULL, client_msg_id VARCHAR(64),
msg_type VARCHAR(32) NOT NULL DEFAULT 'text', status VARCHAR(32) NOT NULL DEFAULT 'normal',
content JSON, content_nonce VARBINARY(12), key_version INT NOT NULL DEFAULT 0,
@@ -45,7 +45,7 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
INDEX idx_im_group_messages_expires(expires_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_members (
id BIGINT NOT NULL AUTO_INCREMENT, topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL,
role VARCHAR(16) NOT NULL DEFAULT 'member', member_status VARCHAR(16) NOT NULL DEFAULT 'active',
nickname VARCHAR(64) NOT NULL DEFAULT '', pinned TINYINT NOT NULL DEFAULT 0,
mute_notifications TINYINT NOT NULL DEFAULT 0, join_source VARCHAR(32) NOT NULL DEFAULT 'manual',
@@ -59,7 +59,7 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
INDEX idx_im_group_members_topic_status(topic, member_status, id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_invitations (
id BIGINT NOT NULL AUTO_INCREMENT, invitation_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, invitation_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
inviter_user_id BIGINT NOT NULL, target_user_id BIGINT NOT NULL, message VARCHAR(256) NOT NULL DEFAULT '',
status VARCHAR(16) NOT NULL DEFAULT 'pending', result VARCHAR(32) NOT NULL DEFAULT '',
join_request_id VARCHAR(64) NOT NULL DEFAULT '', idempotency_key VARCHAR(128) NOT NULL DEFAULT '',
@@ -72,7 +72,7 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
INDEX idx_im_group_invitations_topic_status(topic, status, id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_join_requests (
id BIGINT NOT NULL AUTO_INCREMENT, request_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, request_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
applicant_user_id BIGINT NOT NULL, inviter_user_id BIGINT, source VARCHAR(32) NOT NULL DEFAULT 'manual',
qr_token_hash CHAR(64) NOT NULL DEFAULT '', message VARCHAR(256) NOT NULL DEFAULT '',
status VARCHAR(16) NOT NULL DEFAULT 'pending', reject_reason VARCHAR(256) NOT NULL DEFAULT '',
@@ -87,7 +87,7 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
INDEX idx_im_group_join_requests_expire(status, expires_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_announcements (
id BIGINT NOT NULL AUTO_INCREMENT, announcement_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, announcement_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
author_user_id BIGINT NOT NULL, content TEXT NOT NULL, status VARCHAR(16) NOT NULL DEFAULT 'current',
version INT NOT NULL DEFAULT 1, published_at DATETIME(3) NOT NULL, deleted_by BIGINT, deleted_at DATETIME(3),
idempotency_key VARCHAR(128) NOT NULL DEFAULT '',
@@ -99,14 +99,14 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
INDEX idx_im_group_announcements_topic_time(topic, published_at DESC, id DESC)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_announcement_reads (
id BIGINT NOT NULL, announcement_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL, announcement_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL,
read_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_group_announcement_reads_announcement_user(announcement_id, user_id),
INDEX idx_im_group_announcement_reads_topic_user(topic, user_id, read_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_mutes (
id BIGINT NOT NULL, topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, muted_by BIGINT NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL, topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, muted_by BIGINT NOT NULL,
muted_at DATETIME(3) NOT NULL, expires_at DATETIME(3), reason VARCHAR(256) NOT NULL DEFAULT '',
status VARCHAR(16) NOT NULL DEFAULT 'active', revoked_by BIGINT, revoked_at DATETIME(3),
version INT NOT NULL DEFAULT 1, created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
@@ -116,7 +116,7 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
INDEX idx_im_group_mutes_user_status(user_id, status, expires_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_qrcodes (
id BIGINT NOT NULL AUTO_INCREMENT, token_id VARCHAR(64) NOT NULL, token_hash CHAR(64) NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, token_id VARCHAR(64) NOT NULL, token_hash CHAR(64) NOT NULL,
topic VARCHAR(32) NOT NULL, created_by BIGINT NOT NULL, status VARCHAR(16) NOT NULL DEFAULT 'active',
expires_at DATETIME(3) NOT NULL, revoked_by BIGINT, revoked_at DATETIME(3),
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
@@ -125,7 +125,7 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
UNIQUE INDEX uk_im_group_qrcodes_hash(token_hash), INDEX idx_im_group_qrcodes_topic_status(topic, status, expires_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_files (
id BIGINT NOT NULL AUTO_INCREMENT, group_file_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, group_file_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
file_id VARCHAR(64) NOT NULL, source_topic VARCHAR(32) NOT NULL DEFAULT '', source_seq BIGINT NOT NULL DEFAULT 0,
uploader_id BIGINT NOT NULL, file_name VARCHAR(255) NOT NULL DEFAULT '', mime_type VARCHAR(128) NOT NULL DEFAULT '',
file_type VARCHAR(32) NOT NULL DEFAULT 'file', file_size BIGINT NOT NULL DEFAULT 0,
@@ -140,7 +140,7 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
INDEX idx_im_group_files_topic_name(topic, status, file_name), INDEX idx_im_group_files_uploader(uploader_id, id DESC)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_admin_logs (
id BIGINT NOT NULL AUTO_INCREMENT, log_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, log_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
operator_id BIGINT NOT NULL, target_user_id BIGINT, action VARCHAR(64) NOT NULL,
object_type VARCHAR(32) NOT NULL DEFAULT '', object_id VARCHAR(64) NOT NULL DEFAULT '',
before_data JSON, after_data JSON, request_id VARCHAR(128) NOT NULL DEFAULT '',
@@ -150,7 +150,7 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
INDEX idx_im_group_admin_logs_operator(operator_id, id DESC), INDEX idx_im_group_admin_logs_object(object_type, object_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_idempotency (
id BIGINT NOT NULL AUTO_INCREMENT, user_id BIGINT NOT NULL, method VARCHAR(8) NOT NULL,
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, user_id BIGINT NOT NULL, method VARCHAR(8) NOT NULL,
route VARCHAR(256) NOT NULL, idempotency_key VARCHAR(128) NOT NULL, body_hash CHAR(64) NOT NULL,
result_code INT NOT NULL DEFAULT 0, result_body JSON,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
@@ -168,5 +168,5 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
return err
}
}
return ensureGroupMessageShardInfrastructure(exec)
return ensureGroupMessageShardInfrastructure(exec, defaultTenantID)
}
server/db/mysql/group_schema_test.go
+28 -2
View File
@@ -6,6 +6,8 @@ import (
"database/sql"
"strings"
"testing"
types "github.com/tinode/chat/server/store/types"
)
type recordingGroupSchemaExecutor struct {
@@ -54,12 +56,36 @@ func TestEnsureIMGroupBusinessTablesUsesSelectedTenantDatabase(t *testing.T) {
}
func TestGroupMessageShardTableRejectsUnsafeMonth(t *testing.T) {
if table, ok := groupMessageShardTable("202609"); !ok || table != "im_group_messages_202609" {
if table, ok := groupMessageShardTable(1, "202609"); !ok || table != "im_group_messages_202609" {
t.Fatalf("unexpected valid shard: table=%q ok=%v", table, ok)
}
for _, month := range []string{"", "20269", "202613", "2026_09", "202609;DROP TABLE im_users"} {
if table, ok := groupMessageShardTable(month); ok || table != "" {
if table, ok := groupMessageShardTable(1, month); ok || table != "" {
t.Fatalf("unsafe shard month accepted: month=%q table=%q", month, table)
}
}
}
func TestTenantPartitionTableNames(t *testing.T) {
tests := []struct {
tenantID types.TenantID
friends string
messages string
group string
}{
{tenantID: 1, friends: "im_friendships", messages: "im_messages", group: "im_group_messages_202609"},
{tenantID: 10, friends: "im_friendships", messages: "im_messages", group: "im_group_messages_202609"},
{tenantID: 1000, friends: "im_friendships", messages: "im_messages", group: "im_group_messages_202609"},
}
for _, tc := range tests {
if got := tenantFriendshipsTable(tc.tenantID); got != tc.friends {
t.Fatalf("tenant %d friendship table: got %q, want %q", tc.tenantID, got, tc.friends)
}
if got := tenantMessagesTable(tc.tenantID); got != tc.messages {
t.Fatalf("tenant %d message table: got %q, want %q", tc.tenantID, got, tc.messages)
}
if got, ok := tenantGroupMessagesTable(tc.tenantID, "202609"); !ok || got != tc.group {
t.Fatalf("tenant %d group message table: got %q, ok=%v, want %q", tc.tenantID, got, ok, tc.group)
}
}
}
server/db/mysql/registration_invites.go
+9 -8
View File
@@ -5,6 +5,7 @@ package mysql
import (
"database/sql"
"fmt"
"strconv"
"strings"
"time"
@@ -390,7 +391,7 @@ func (a *adapter) TenantRegistrationInviteRecordUse(id t.TenantID, value string,
if row.MaxUses > 0 {
var usedCount int
if err = tx.GetContext(txCtx, &usedCount,
"SELECT COUNT(*) FROM im_registration_invite_uses WHERE invite_code_id=?", row.ID); err != nil {
"SELECT COUNT(*) FROM im_registration_invite_uses WHERE tenant_id=? AND invite_code_id=?", id, row.ID); err != nil {
return err
}
if usedCount >= row.MaxUses {
@@ -402,12 +403,12 @@ func (a *adapter) TenantRegistrationInviteRecordUse(id t.TenantID, value string,
return err
}
if _, err = tx.ExecContext(txCtx, `INSERT INTO im_registration_invite_uses
(id,invite_code_id,user_id,request_id,used_at) VALUES(?,?,?,?,?)`, useID, row.ID,
(tenant_id,id,invite_code_id,user_id,request_id,used_at) VALUES(?,?,?,?,?,?)`, id, useID, row.ID,
store.DecodeUid(uid), requestID, now); err != nil {
if isDupe(err) {
var existingCodeID int64
if lookupErr := tx.GetContext(txCtx, &existingCodeID,
"SELECT invite_code_id FROM im_registration_invite_uses WHERE user_id=?", store.DecodeUid(uid)); lookupErr == nil && existingCodeID == row.ID {
"SELECT invite_code_id FROM im_registration_invite_uses WHERE tenant_id=? AND user_id=?", id, store.DecodeUid(uid)); lookupErr == nil && existingCodeID == row.ID {
err = nil
} else {
return t.ErrDuplicate
@@ -419,7 +420,7 @@ func (a *adapter) TenantRegistrationInviteRecordUse(id t.TenantID, value string,
if row.BoundUserID != nil && *row.BoundUserID > 0 {
boundUID := t.ParseUserId(strconv.FormatInt(*row.BoundUserID, 10))
if !boundUID.IsZero() && boundUID != uid {
if err = upsertFriendshipPairTx(tx, uid, boundUID, uid.P2PName(boundUID), now); err != nil {
if err = upsertFriendshipPairTx(tx, id, uid, boundUID, uid.P2PName(boundUID), now); err != nil {
return err
}
}
@@ -430,7 +431,7 @@ func (a *adapter) TenantRegistrationInviteRecordUse(id t.TenantID, value string,
return a.TenantRegistrationInviteSyncUsage(id, row.ID)
}
func upsertFriendshipPairTx(tx *sqlx.Tx, uid1, uid2 t.Uid, topic string, now time.Time) error {
func upsertFriendshipPairTx(tx *sqlx.Tx, tenantID t.TenantID, uid1, uid2 t.Uid, topic string, now time.Time) error {
if uid1.IsZero() || uid2.IsZero() || uid1 == uid2 {
return t.ErrMalformed
}
@@ -443,9 +444,9 @@ func upsertFriendshipPairTx(tx *sqlx.Tx, uid1, uid2 t.Uid, topic string, now tim
return err
}
if _, err = tx.Exec(
"INSERT INTO im_friendships(id,user_id,peer_id,status,topic,created_at,updated_at) VALUES(?,?,?,?,?,?,?) "+
fmt.Sprintf("INSERT INTO %s(tenant_id,id,user_id,peer_id,status,topic,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?) ", tenantFriendshipsTable(tenantID))+
"ON DUPLICATE KEY UPDATE status=VALUES(status),topic=VALUES(topic),updated_at=VALUES(updated_at)",
friendshipID, decodeUID(pair[0]), decodeUID(pair[1]), t.RelationshipFriend, topic, now, now); err != nil {
tenantID, friendshipID, decodeUID(pair[0]), decodeUID(pair[1]), t.RelationshipFriend, topic, now, now); err != nil {
return err
}
}
@@ -474,7 +475,7 @@ func (a *adapter) TenantRegistrationInviteSyncUsage(id t.TenantID, codeID int64)
LastUsedAt *time.Time `db:"last_used_at"`
}
if err = a.db.GetContext(ctx, &aggregate, `SELECT COUNT(*) AS usage_count,MAX(used_at) AS last_used_at
FROM im_registration_invite_uses WHERE invite_code_id=?`, codeID); err != nil {
FROM im_registration_invite_uses WHERE tenant_id=? AND invite_code_id=?`, id, codeID); err != nil {
return err
}
res, err := a.registrationInviteControlDB().ExecContext(ctx, `UPDATE im_registration_invite_codes
server/db/mysql/report_group_regression_test.go
+1 -1
View File
@@ -158,7 +158,7 @@ func TestTenantFileAccessIncludesAvatarAudienceRules(t *testing.T) {
}
body := string(source)
for _, want := range []string{
"INNER JOIN im_friendships AS fs",
"INNER JOIN %s AS fs",
"fs.user_id=? AND fs.peer_id=fml.userid AND fs.status=?",
"INNER JOIN im_group_members AS gm",
"gm.topic=fml.topic AND gm.user_id=? AND gm.member_status=?",
server/db/mysql/tenant_adapter.go
+443 -384
View File
File diff suppressed because it is too large Load Diff
server/db/mysql/tenant_infrastructure.go
+17 -16
View File
@@ -42,8 +42,8 @@ func (a *adapter) TenantHTTPIdempotencyBegin(id t.TenantID, record *t.HTTPIdempo
record.ExpiresAt = now.Add(24 * time.Hour)
}
_, err = a.db.Exec(`INSERT INTO im_http_idempotency
(user_id,method,route,idempotency_key,body_hash,state,response_status,response_body,expires_at,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?)`, decodeUID(record.UserID), record.Method, record.Route, record.Key,
(tenant_id,user_id,method,route,idempotency_key,body_hash,state,response_status,response_body,expires_at,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, id, decodeUID(record.UserID), record.Method, record.Route, record.Key,
record.BodyHash, t.HTTPIdempotencyProcessing, 0, nil, record.ExpiresAt, now, now)
if err == nil {
record.State = t.HTTPIdempotencyProcessing
@@ -54,18 +54,18 @@ func (a *adapter) TenantHTTPIdempotencyBegin(id t.TenantID, record *t.HTTPIdempo
if !isDupe(err) {
return nil, false, err
}
existing, err := a.httpIdempotencyGet(record.UserID, record.Method, record.Route, record.Key)
existing, err := a.httpIdempotencyGet(id, record.UserID, record.Method, record.Route, record.Key)
if err == nil && !existing.ExpiresAt.After(now) {
res, deleteErr := a.db.Exec(`DELETE FROM im_http_idempotency
WHERE user_id=? AND method=? AND route=? AND idempotency_key=? AND expires_at<=?`,
decodeUID(record.UserID), record.Method, record.Route, record.Key, now)
WHERE tenant_id=? AND user_id=? AND method=? AND route=? AND idempotency_key=? AND expires_at<=?`,
id, decodeUID(record.UserID), record.Method, record.Route, record.Key, now)
if deleteErr != nil {
return nil, false, deleteErr
}
if deleted, _ := res.RowsAffected(); deleted > 0 {
return a.TenantHTTPIdempotencyBegin(id, record)
}
current, getErr := a.httpIdempotencyGet(record.UserID, record.Method, record.Route, record.Key)
current, getErr := a.httpIdempotencyGet(id, record.UserID, record.Method, record.Route, record.Key)
return current, false, getErr
}
return existing, false, err
@@ -84,8 +84,8 @@ func (a *adapter) TenantHTTPIdempotencyComplete(id t.TenantID, record *t.HTTPIde
}
now := t.TimeNow()
res, err := a.db.Exec(`UPDATE im_http_idempotency SET state=?,response_status=?,response_body=?,updated_at=?
WHERE user_id=? AND method=? AND route=? AND idempotency_key=? AND body_hash=? AND state=?`,
t.HTTPIdempotencyCompleted, record.ResponseStatus, record.ResponseBody, now, decodeUID(record.UserID),
WHERE tenant_id=? AND user_id=? AND method=? AND route=? AND idempotency_key=? AND body_hash=? AND state=?`,
t.HTTPIdempotencyCompleted, record.ResponseStatus, record.ResponseBody, now, id, decodeUID(record.UserID),
record.Method, record.Route, record.Key, record.BodyHash, t.HTTPIdempotencyProcessing)
if err != nil {
return err
@@ -108,12 +108,12 @@ func (a *adapter) TenantHTTPIdempotencyAbort(id t.TenantID, record *t.HTTPIdempo
return tenantAdp.TenantHTTPIdempotencyAbort(id, record)
}
_, err = a.db.Exec(`DELETE FROM im_http_idempotency
WHERE user_id=? AND method=? AND route=? AND idempotency_key=? AND body_hash=? AND state=?`,
decodeUID(record.UserID), record.Method, record.Route, record.Key, record.BodyHash, t.HTTPIdempotencyProcessing)
WHERE tenant_id=? AND user_id=? AND method=? AND route=? AND idempotency_key=? AND body_hash=? AND state=?`,
id, decodeUID(record.UserID), record.Method, record.Route, record.Key, record.BodyHash, t.HTTPIdempotencyProcessing)
return err
}
func (a *adapter) httpIdempotencyGet(uid t.Uid, method, route, key string) (*t.HTTPIdempotencyRecord, error) {
func (a *adapter) httpIdempotencyGet(tenantID t.TenantID, uid t.Uid, method, route, key string) (*t.HTTPIdempotencyRecord, error) {
row := struct {
UserID int64 `db:"user_id"`
Method string
@@ -128,7 +128,8 @@ func (a *adapter) httpIdempotencyGet(uid t.Uid, method, route, key string) (*t.H
ExpiresAt time.Time `db:"expires_at"`
}{}
err := a.db.Get(&row, `SELECT user_id,method,route,idempotency_key,body_hash,state,response_status,response_body,created_at,updated_at,expires_at
FROM im_http_idempotency WHERE user_id=? AND method=? AND route=? AND idempotency_key=?`, decodeUID(uid), method, route, key)
FROM im_http_idempotency WHERE tenant_id=? AND user_id=? AND method=? AND route=? AND idempotency_key=?`,
tenantID, decodeUID(uid), method, route, key)
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}
@@ -512,16 +513,16 @@ func truncateKafkaString(value string, limit int) string {
return string(runes[:limit])
}
func insertSystemNotificationTx(tx *sqlx.Tx, uid t.Uid, eventID, eventType, title, summary string, action any) error {
func insertSystemNotificationTx(tx *sqlx.Tx, tenantID t.TenantID, uid t.Uid, eventID, eventType, title, summary string, action any) error {
encoded, err := json.Marshal(action)
if err != nil {
return err
}
now := t.TimeNow()
_, err = tx.Exec(`INSERT INTO im_system_notifications
(user_id,event_id,event_type,title,summary,action,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE event_type=VALUES(event_type),title=VALUES(title),
summary=VALUES(summary),action=VALUES(action),updated_at=VALUES(updated_at)`, decodeUID(uid), eventID,
(tenant_id,user_id,event_id,event_type,title,summary,action,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE event_type=VALUES(event_type),title=VALUES(title),
summary=VALUES(summary),action=VALUES(action),updated_at=VALUES(updated_at)`, tenantID, decodeUID(uid), eventID,
eventType, title, summary, encoded, now, now)
return err
}
server/db/mysql/tenant_security_audit.go
+4 -3
View File
@@ -134,7 +134,7 @@ func (a *adapter) TenantMessageAdminApply(id t.TenantID, action *t.MessageAdminA
if action.ConversationType == "group" {
_, err = tx.ExecContext(ctx, "UPDATE "+messageTable+" SET status=?,deleted_at=?,updated_at=? WHERE topic=? AND seqid=?", afterStatus, action.ProcessedAt, action.ProcessedAt, action.Topic, action.SeqID)
} else {
_, err = tx.ExecContext(ctx, "UPDATE im_messages SET deletedat=?,updatedat=? WHERE topic=? AND seqid=?", action.ProcessedAt, action.ProcessedAt, action.Topic, action.SeqID)
_, err = tx.ExecContext(ctx, "UPDATE "+messageTable+" SET deletedat=?,updatedat=? WHERE topic=? AND seqid=?", action.ProcessedAt, action.ProcessedAt, action.Topic, action.SeqID)
}
if err != nil {
return err
@@ -177,7 +177,8 @@ func lockAdminMessage(ctx context.Context, tx *sqlx.Tx, tenantID t.TenantID, con
DeletedAt sql.NullTime `db:"deletedat"`
RecalledAt sql.NullTime `db:"recalled_at"`
}
err := tx.GetContext(ctx, &row, "SELECT `from`,deletedat,recalled_at FROM im_messages WHERE topic=? AND seqid=? FOR UPDATE", topic, seqID)
messageTable := tenantMessagesTable(tenantID)
err := tx.GetContext(ctx, &row, "SELECT `from`,deletedat,recalled_at FROM "+messageTable+" WHERE topic=? AND seqid=? FOR UPDATE", topic, seqID)
if errors.Is(err, sql.ErrNoRows) {
return 0, "", "", t.ErrNotFound
}
@@ -188,7 +189,7 @@ func lockAdminMessage(ctx context.Context, tx *sqlx.Tx, tenantID t.TenantID, con
if row.DeletedAt.Valid {
status = "deleted"
}
return row.Sender, status, "im_messages", err
return row.Sender, status, messageTable, err
}
func nullableJSON(raw []byte) any {
server/db/mysql/tenant_tables.go
+37
View File
@@ -0,0 +1,37 @@
//go:build mysql
// +build mysql
package mysql
import (
"fmt"
t "github.com/tinode/chat/server/store/types"
)
func tenantFriendshipsTable(t.TenantID) string {
return "im_friendships"
}
func tenantMessagesTable(t.TenantID) string {
return "im_messages"
}
func tenantGroupMessagesTable(_ t.TenantID, month string) (string, bool) {
if !validShardMonth(month) {
return "", false
}
return fmt.Sprintf("im_group_messages_%s", month), true
}
func validShardMonth(month string) bool {
if len(month) != 6 {
return false
}
for i := range month {
if month[i] < '0' || month[i] > '9' {
return false
}
}
return month[4:] >= "01" && month[4:] <= "12"
}
server/db/mysql/tests/mysql_test.go
+6 -6
View File
@@ -84,8 +84,8 @@ func TestGroupBusinessTablesCreated(t *testing.T) {
if err := db.Get(&count, query, args...); err != nil {
t.Fatal(err)
}
if count != 0 {
t.Fatalf("tenant business tables contain %d tenant_id columns", count)
if count != len(tables) {
t.Fatalf("tenant business tables with tenant_id: got %d, want %d", count, len(tables))
}
}
@@ -264,12 +264,12 @@ func TestTenantAuthIsolation(t *testing.T) {
}
}
createTenant := func(code, dbName string) types.TenantID {
res, err := db.Exec("INSERT INTO im_tenant(code,name,db_name,state,created_by,updated_by) VALUES(?,?,?,?,?,?)",
code+suffix, code, dbName, types.TenantStateActive, 0, 0)
if err != nil {
var id int64
if err := db.Get(&id, "SELECT COALESCE(MAX(id),0)+1 FROM im_tenant"); err != nil {
t.Fatal(err)
}
id, err := res.LastInsertId()
_, err := db.Exec("INSERT INTO im_tenant(id,code,name,db_name,state,created_by,updated_by) VALUES(?,?,?,?,?,?,?)",
id, code+suffix, code, dbName, types.TenantStateActive, 0, 0)
if err != nil {
t.Fatal(err)
}