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

fix shared-db tenant runtime gaps

Browse Source
This commit is contained in:
richjohnrambo
2026-09-22 19:28:33 +08:00
parent 49e43aa3cf
commit 9ddf6ec5ee
12 changed files with 346 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/infrastructure_schema.go
+12 -2
View File
@@ -6,7 +6,7 @@ package mysql
func ensureIMHTTPInfrastructureTables(exec sqlExecutor) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS im_http_idempotency (
id BIGINT NOT NULL AUTO_INCREMENT, user_id BIGINT NOT NULL,
id BIGINT NOT NULL AUTO_INCREMENT, tenant_id BIGINT NOT NULL DEFAULT 0, user_id BIGINT NOT NULL,
method VARCHAR(8) NOT NULL, route VARCHAR(512) NOT NULL,
idempotency_key VARCHAR(128) NOT NULL, body_hash CHAR(64) NOT NULL,
state VARCHAR(16) NOT NULL DEFAULT 'processing', response_status INT NOT NULL DEFAULT 0,
@@ -14,7 +14,7 @@ func ensureIMHTTPInfrastructureTables(exec sqlExecutor) error {
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_http_idempotency_scope(user_id, method, route, idempotency_key),
UNIQUE INDEX uk_im_http_idempotency_scope(tenant_id, user_id, method, route, idempotency_key),
INDEX idx_im_http_idempotency_expiry(expires_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_kafka_consume_log (
@@ -82,5 +82,15 @@ func ensureIMHTTPInfrastructureTables(exec sqlExecutor) error {
return err
}
}
migrations := []ignoreSQLError{
{`ALTER TABLE im_http_idempotency ADD COLUMN tenant_id BIGINT NOT NULL DEFAULT 0 AFTER id`, 1060},
{`ALTER TABLE im_http_idempotency DROP INDEX uk_im_http_idempotency_scope`, 1091},
{`ALTER TABLE im_http_idempotency ADD UNIQUE INDEX uk_im_http_idempotency_scope(tenant_id, user_id, method, route, idempotency_key)`, 1061},
}
for _, migration := range migrations {
if err := execIgnoreMySQLError(exec, migration.sql, migration.code); err != nil {
return err
}
}
return nil
}
server/db/mysql/infrastructure_schema_test.go
+6 -3
View File
@@ -12,8 +12,8 @@ func TestEnsureIMHTTPInfrastructureTablesUsesSelectedTenantDatabase(t *testing.T
if err := ensureIMHTTPInfrastructureTables(exec); err != nil {
t.Fatal(err)
}
if len(exec.statements) != 6 {
t.Fatalf("created %d infrastructure tables, want 6", len(exec.statements))
if len(exec.statements) != 9 {
t.Fatalf("executed %d infrastructure schema statements, want 9", len(exec.statements))
}
joined := strings.ToLower(strings.Join(exec.statements, "\n"))
for _, table := range []string{
@@ -28,9 +28,12 @@ func TestEnsureIMHTTPInfrastructureTablesUsesSelectedTenantDatabase(t *testing.T
t.Fatalf("missing table %s", table)
}
}
if strings.Contains(joined, "tinode.") || strings.Contains(joined, "create database") || strings.Contains(joined, "tenant_id") {
if strings.Contains(joined, "tinode.") || strings.Contains(joined, "create database") {
t.Fatalf("infrastructure schema must stay inside the selected tenant database: %s", joined)
}
if !strings.Contains(joined, "tenant_id") || !strings.Contains(joined, "uk_im_http_idempotency_scope(tenant_id, user_id") {
t.Fatalf("HTTP idempotency schema must be tenant scoped in shared DB mode: %s", joined)
}
if strings.Contains(joined, "im_kafka_outbox") {
t.Fatalf("kafka outbox table must not be created: %s", joined)
}
server/db/mysql/tenant_adapter.go
+78 -6
View File
@@ -897,6 +897,78 @@ func tenantTopicCreate(tx *sqlx.Tx, topic *t.Topic) error {
return addTenantTags(tx, topic.TenantID, "im_topictags", "topic", topic.Id, topic.Tags, false)
}
// createTenantSubscription mirrors createSubscription but keeps the subscription
// bound to the logical tenant in shared-database deployments.
func createTenantSubscription(tx *sqlx.Tx, tenantID t.TenantID, sub *t.Subscription, undelete bool) error {
if sub == nil || tenantID.IsZero() || sub.TenantID != tenantID {
return t.ErrMalformed
}
isOwner := (sub.ModeGiven & sub.ModeWant).IsOwner()
jpriv := common.ToJSON(sub.Private)
decodedUID := decodeUID(t.ParseUid(sub.User))
existing := struct {
ID int64 `db:"id"`
DeletedAt sql.NullTime `db:"deletedat"`
}{}
err := tx.Get(&existing,
"SELECT id,deletedat FROM im_subscriptions WHERE tenant_id=? AND topic=? AND userid=? ORDER BY deletedat IS NULL DESC,id ASC LIMIT 1",
tenantID, sub.Topic, decodedUID)
if err != nil && err != sql.ErrNoRows {
return err
}
if err == nil {
if existing.DeletedAt.Valid {
_, err = tx.Exec("UPDATE im_subscriptions SET createdat=?,updatedat=?,deletedat=NULL,modeWant=?,modeGiven=?,"+
"delid=0,recvseqid=0,readseqid=0,private=? WHERE tenant_id=? AND id=?",
sub.CreatedAt, sub.UpdatedAt, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv, tenantID, existing.ID)
} else {
_, err = tx.Exec("UPDATE im_subscriptions SET updatedat=?,modeWant=?,modeGiven=?,private=? WHERE tenant_id=? AND id=?",
sub.UpdatedAt, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv, tenantID, existing.ID)
}
if err != nil {
return err
}
if isOwner {
_, err = tx.Exec("UPDATE im_topics SET owner=? WHERE tenant_id=? AND name=?", decodedUID, tenantID, sub.Topic)
}
return err
}
subID, err := nextSnowflakeID()
if err != nil {
return err
}
_, err = tx.Exec(
"INSERT INTO im_subscriptions(id,tenant_id,createdat,updatedat,deletedat,userid,topic,modeWant,modeGiven,private) "+
"VALUES(?,?,?,?,NULL,?,?,?,?,?)",
subID, tenantID, sub.CreatedAt, sub.UpdatedAt, decodedUID, sub.Topic, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv)
if isIDOutOfRange(err) {
_, err = tx.Exec(
"INSERT INTO im_subscriptions(tenant_id,createdat,updatedat,deletedat,userid,topic,modeWant,modeGiven,private) "+
"VALUES(?,?,?,NULL,?,?,?,?,?)",
tenantID, sub.CreatedAt, sub.UpdatedAt, decodedUID, sub.Topic, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv)
}
if err != nil && isDupe(err) {
if undelete {
_, err = tx.Exec("UPDATE im_subscriptions SET createdat=?,updatedat=?,deletedat=NULL,modeWant=?,modeGiven=?,"+
"delid=0,recvseqid=0,readseqid=0 WHERE tenant_id=? AND topic=? AND userid=?",
sub.CreatedAt, sub.UpdatedAt, sub.ModeWant.String(), sub.ModeGiven.String(), tenantID, sub.Topic, decodedUID)
} else {
_, err = tx.Exec("UPDATE im_subscriptions SET createdat=?,updatedat=?,deletedat=NULL,modeWant=?,modeGiven=?,"+
"delid=0,recvseqid=0,readseqid=0,private=? WHERE tenant_id=? AND topic=? AND userid=?",
sub.CreatedAt, sub.UpdatedAt, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv, tenantID, sub.Topic, decodedUID)
}
}
if err == nil && isOwner {
_, err = tx.Exec("UPDATE im_topics SET owner=? WHERE tenant_id=? AND name=?", decodedUID, tenantID, sub.Topic)
}
return err
}
func (a *adapter) TenantTopicCreate(id t.TenantID, topic *t.Topic) error {
tenantAdp, err := a.routedTenantAdapter(id)
if err != nil || topic == nil || topic.TenantID != id {
@@ -951,17 +1023,17 @@ func (a *adapter) TenantTopicCreateP2P(id t.TenantID, one, two *t.Subscription)
tx.Rollback()
}
}()
if err = createSubscription(tx, one, false); err != nil {
if err = createTenantSubscription(tx, id, one, false); err != nil {
return err
}
if err = createSubscription(tx, two, true); err != nil {
if err = createTenantSubscription(tx, id, two, true); err != nil {
return err
}
topic := &t.Topic{TenantID: id, ObjHeader: t.ObjHeader{Id: one.Topic}}
topic.ObjHeader.MergeTimes(&one.ObjHeader)
topic.TouchedAt = one.GetTouchedAt()
var existingTopicID int64
err = tx.Get(&existingTopicID, "SELECT id FROM im_topics WHERE name=? ORDER BY id ASC LIMIT 1", topic.Id)
err = tx.Get(&existingTopicID, "SELECT id FROM im_topics WHERE tenant_id=? AND name=? ORDER BY id ASC LIMIT 1", id, topic.Id)
if err != nil && err != sql.ErrNoRows {
return err
}
@@ -969,8 +1041,8 @@ func (a *adapter) TenantTopicCreateP2P(id t.TenantID, one, two *t.Subscription)
if err = tenantTopicCreate(tx, topic); err != nil {
return err
}
} else if _, err = tx.ExecContext(ctx, "UPDATE im_topics SET updatedat=?,touchedat=?,state=? WHERE id=?",
topic.UpdatedAt, topic.TouchedAt, t.StateOK, existingTopicID); err != nil {
} else if _, err = tx.ExecContext(ctx, "UPDATE im_topics SET updatedat=?,touchedat=?,state=? WHERE tenant_id=? AND id=?",
topic.UpdatedAt, topic.TouchedAt, t.StateOK, id, existingTopicID); err != nil {
return err
}
return tx.Commit()
@@ -1403,7 +1475,7 @@ func (a *adapter) TenantTopicShare(id t.TenantID, topic string, subs []*t.Subscr
}
}()
for _, sub := range subs {
if err = createSubscription(tx, sub, true); err != nil {
if err = createTenantSubscription(tx, id, sub, true); err != nil {
return err
}
}
server/db/mysql/tests/mysql_test.go
+70 -2
View File
@@ -112,8 +112,8 @@ func TestHTTPInfrastructureTablesCreated(t *testing.T) {
if err := db.Get(&count, query, args...); err != nil {
t.Fatal(err)
}
if count != 0 {
t.Fatalf("HTTP infrastructure tables contain %d tenant_id columns", count)
if count != len(tables) {
t.Fatalf("HTTP infrastructure tenant_id columns: got %d, want %d", count, len(tables))
}
}
@@ -644,6 +644,74 @@ func TestTopicCreateP2P(t *testing.T) {
}
}
func TestTenantTopicCreateP2PWritesTenantSubscriptions(t *testing.T) {
tenantAdapter, ok := adp.(adapter.TenantBusinessAdapter)
if !ok {
t.Fatal("MySQL adapter does not implement TenantBusinessAdapter")
}
var tenantID types.TenantID
if err := db.Get(&tenantID, "SELECT id FROM im_tenant WHERE code='default'"); err != nil {
t.Fatal(err)
}
var tenantColumns int
if err := db.Get(&tenantColumns, `SELECT COUNT(*) FROM information_schema.columns
WHERE table_schema=DATABASE() AND table_name='im_subscriptions' AND column_name='tenant_id'`); err != nil {
t.Fatal(err)
}
if tenantColumns == 0 {
t.Skip("im_subscriptions.tenant_id is missing in the current test database")
}
first := types.ParseUserId("usr" + testData.Users[0].Id)
second := types.ParseUserId("usr" + testData.Users[2].Id)
topic := first.P2PName(second)
if _, err := db.Exec("DELETE FROM im_subscriptions WHERE tenant_id=? AND topic=?", tenantID, topic); err != nil {
t.Fatal(err)
}
if _, err := db.Exec("DELETE FROM im_topics WHERE tenant_id=? AND name=?", tenantID, topic); err != nil {
t.Fatal(err)
}
now := types.TimeNow()
one := &types.Subscription{
TenantID: tenantID,
ObjHeader: types.ObjHeader{CreatedAt: now, UpdatedAt: now},
User: first.String(),
Topic: topic,
ModeWant: types.ModeCPublic,
ModeGiven: types.ModeCPublic,
}
one.SetTouchedAt(now)
two := &types.Subscription{
TenantID: tenantID,
ObjHeader: types.ObjHeader{CreatedAt: now, UpdatedAt: now},
User: second.String(),
Topic: topic,
ModeWant: types.ModeCPublic,
ModeGiven: types.ModeCPublic,
}
two.SetTouchedAt(now)
if err := tenantAdapter.TenantTopicCreateP2P(tenantID, one, two); err != nil {
t.Fatal(err)
}
var count int
if err := db.Get(&count, "SELECT COUNT(*) FROM im_subscriptions WHERE tenant_id=? AND topic=?", tenantID, topic); err != nil {
t.Fatal(err)
}
if count != 2 {
t.Fatalf("tenant P2P subscriptions: got %d, want 2", count)
}
subs, err := tenantAdapter.TenantUsersForTopic(tenantID, topic, false, nil)
if err != nil {
t.Fatal(err)
}
if len(subs) != 2 {
t.Fatalf("TenantUsersForTopic returned %d subscriptions, want 2", len(subs))
}
}
func TestTopicShare(t *testing.T) {
if err := adp.TopicShare(testData.Subs[0].Topic, testData.Subs); err != nil {
t.Fatal(err)
server/db/mysql/user_identity_schema.go
+46 -24
View File
@@ -31,26 +31,6 @@ func ensureUserIdentitySchema(db *sqlx.DB) error {
}
defer conn.Close()
// Fast path: the last two steps of this migration drop the legacy im_usertags
// table and the im_users.tags column. If both are already gone, the migration
// has fully run (or the database was created fresh) and no lock is needed.
stale := false
if exists, err := identityTableExists(ctx, conn, "im_usertags"); err != nil {
return err
} else if exists {
stale = true
}
if !stale {
if exists, err := identityColumnExists(ctx, conn, "im_users", "tags"); err != nil {
return err
} else if exists {
stale = true
}
}
if !stale {
return nil
}
// PXC strict-mode clusters prohibit GET_LOCK; acquireMigrationLock falls back
// to lock-free execution in that case. The migration body is idempotent.
_, unlock, err := acquireMigrationLock(ctx, conn, userIdentityMigrationLock)
@@ -112,7 +92,7 @@ func ensureUserIdentitySchema(db *sqlx.DB) error {
return err
}
}
if err = ensureUserAccountNoSchema(ctx, conn); err != nil {
if err = ensureUserAccountNoSchema(ctx, conn, usersScoped); err != nil {
return err
}
if profilesExist {
@@ -173,6 +153,15 @@ func ensureUserIdentitySchema(db *sqlx.DB) error {
return fmt.Errorf("create unique phone index: %w", err)
}
}
if usersScoped {
if exists, err := identityIndexExists(ctx, conn, "im_users", "users_phone"); err != nil {
return err
} else if exists {
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_users DROP INDEX users_phone"); err != nil {
return err
}
}
}
usernameIndex := "users_username"
usernameColumns := "username"
if usersScoped {
@@ -186,6 +175,15 @@ func ensureUserIdentitySchema(db *sqlx.DB) error {
return fmt.Errorf("create unique username index: %w", err)
}
}
if usersScoped {
if exists, err := identityIndexExists(ctx, conn, "im_users", "users_username"); err != nil {
return err
} else if exists {
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_users DROP INDEX users_username"); err != nil {
return err
}
}
}
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_users MODIFY COLUMN username VARCHAR(32) NOT NULL"); err != nil {
return err
}
@@ -376,6 +374,15 @@ func migrateBasicAuthSchema(ctx context.Context, conn *sqlx.Conn, scoped bool) e
}
}
}
if scoped {
if exists, err = identityIndexExists(ctx, conn, "im_auth", "auth_userid"); err != nil {
return err
} else if exists {
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_auth DROP INDEX auth_userid"); err != nil {
return err
}
}
}
return nil
}
@@ -559,7 +566,7 @@ func identityCandidates(ctx context.Context, conn *sqlx.Conn, scoped bool, metho
return out, nil
}
func ensureUserAccountNoSchema(ctx context.Context, conn *sqlx.Conn) error {
func ensureUserAccountNoSchema(ctx context.Context, conn *sqlx.Conn, scoped bool) error {
if exists, err := identityColumnExists(ctx, conn, "im_users", "account_no"); err != nil {
return err
} else if !exists {
@@ -582,13 +589,28 @@ func ensureUserAccountNoSchema(ctx context.Context, conn *sqlx.Conn) error {
WHERE u.account_no=0`, start); err != nil {
return err
}
if exists, err := identityIndexExists(ctx, conn, "im_users", "uk_im_users_account_no"); err != nil {
indexName := "uk_im_users_account_no"
columns := "account_no"
if scoped {
indexName = "uk_im_users_tenant_account_no"
columns = "tenant_id,account_no"
}
if exists, err := identityIndexExists(ctx, conn, "im_users", indexName); err != nil {
return err
} else if !exists {
if _, err = conn.ExecContext(ctx, "CREATE UNIQUE INDEX uk_im_users_account_no ON im_users(account_no)"); err != nil {
if _, err = conn.ExecContext(ctx, "CREATE UNIQUE INDEX "+quoteIdent(indexName)+" ON im_users("+columns+")"); err != nil {
return fmt.Errorf("create unique user account number index: %w", err)
}
}
if scoped {
if exists, err := identityIndexExists(ctx, conn, "im_users", "uk_im_users_account_no"); err != nil {
return err
} else if exists {
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_users DROP INDEX uk_im_users_account_no"); err != nil {
return err
}
}
}
return nil
}