From 9ddf6ec5eeef6dcb170dbb95c363c39bba5f69e6 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Tue, 22 Sep 2026 19:28:05 +0800 Subject: [PATCH] fix shared-db tenant runtime gaps --- docker/tinode/Dockerfile | 2 + docker/tinode/Dockerfile.build | 2 + docker/tinode/config.template | 34 ++++---- docs/IM-数据库变更汇总.sql | 3 +- docs/sql/IM-共库运行态缺口修复SQL.sql | 59 +++++++++++++ server/db/mysql/infrastructure_schema.go | 14 +++- server/db/mysql/infrastructure_schema_test.go | 9 +- server/db/mysql/tenant_adapter.go | 84 +++++++++++++++++-- server/db/mysql/tests/mysql_test.go | 72 +++++++++++++++- server/db/mysql/user_identity_schema.go | 70 ++++++++++------ server/im_http.go | 31 ++++++- server/im_http_test.go | 24 ++++++ 12 files changed, 346 insertions(+), 58 deletions(-) diff --git a/docker/tinode/Dockerfile b/docker/tinode/Dockerfile index b637831..301df6f 100644 --- a/docker/tinode/Dockerfile +++ b/docker/tinode/Dockerfile @@ -53,6 +53,8 @@ ENV DEFAULT_COUNTRY_CODE=US # The MySQL DSN connection. ENV MYSQL_DSN='root@tcp(mysql)/meeting_im?parseTime=true&collation=utf8mb4_0900_ai_ci' ENV REDIS_ADDR= +ENV REDIS_SENTINEL_MASTER_NAME= +ENV REDIS_SENTINEL_ADDRS_JSON='[]' ENV REDIS_USERNAME= ENV REDIS_PASSWORD= ENV REDIS_DB=0 diff --git a/docker/tinode/Dockerfile.build b/docker/tinode/Dockerfile.build index ab6464f..d0f6db6 100644 --- a/docker/tinode/Dockerfile.build +++ b/docker/tinode/Dockerfile.build @@ -35,6 +35,8 @@ ENV DEFAULT_COUNTRY_CODE=US ENV STORE_USE_ADAPTER=mysql ENV REDIS_ADDR= +ENV REDIS_SENTINEL_MASTER_NAME= +ENV REDIS_SENTINEL_ADDRS_JSON='[]' ENV REDIS_USERNAME= ENV REDIS_DB=0 ENV REDIS_DIAL_TIMEOUT_MS=2000 diff --git a/docker/tinode/config.template b/docker/tinode/config.template index a0d541b..0c0b2c8 100644 --- a/docker/tinode/config.template +++ b/docker/tinode/config.template @@ -101,22 +101,24 @@ } }, "redis": { - "addr": "redis.118im.com:30637", - "username": "", - "password": "HMTJzkm.h>Y+E9ep", - "db": 0, - "key_prefix": "im", - "dial_timeout_ms": 2000, - "read_timeout_ms": 2000, - "write_timeout_ms": 2000, - "pending_ttl_seconds": 300, - "done_ttl_seconds": 86400, - "failed_ttl_seconds": 600, - "hot_message_ttl_seconds": 86400, - "presence_ttl_seconds": 120, - "cache_ttl_seconds": 300, - "short_state_ttl_seconds": 600, - "group_fanout_enabled": true + "addr": "$REDIS_ADDR", + "sentinel_master_name": "$REDIS_SENTINEL_MASTER_NAME", + "sentinel_addrs": $REDIS_SENTINEL_ADDRS_JSON, + "username": "$REDIS_USERNAME", + "password": "$REDIS_PASSWORD", + "db": $REDIS_DB, + "key_prefix": "$REDIS_KEY_PREFIX", + "dial_timeout_ms": $REDIS_DIAL_TIMEOUT_MS, + "read_timeout_ms": $REDIS_READ_TIMEOUT_MS, + "write_timeout_ms": $REDIS_WRITE_TIMEOUT_MS, + "pending_ttl_seconds": $REDIS_PENDING_TTL_SECONDS, + "done_ttl_seconds": $REDIS_DONE_TTL_SECONDS, + "failed_ttl_seconds": $REDIS_FAILED_TTL_SECONDS, + "hot_message_ttl_seconds": $REDIS_HOT_MESSAGE_TTL_SECONDS, + "presence_ttl_seconds": $REDIS_PRESENCE_TTL_SECONDS, + "cache_ttl_seconds": $REDIS_CACHE_TTL_SECONDS, + "short_state_ttl_seconds": $REDIS_SHORT_STATE_TTL_SECONDS, + "group_fanout_enabled": $REDIS_GROUP_FANOUT_ENABLED } }, diff --git a/docs/IM-数据库变更汇总.sql b/docs/IM-数据库变更汇总.sql index d8df866..4059e68 100644 --- a/docs/IM-数据库变更汇总.sql +++ b/docs/IM-数据库变更汇总.sql @@ -166,6 +166,7 @@ CREATE TABLE IF NOT EXISTS tinode.im_system_notifications ( CREATE TABLE IF NOT EXISTS tinode.im_http_idempotency ( 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, @@ -178,7 +179,7 @@ CREATE TABLE IF NOT EXISTS tinode.im_http_idempotency ( 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; diff --git a/docs/sql/IM-共库运行态缺口修复SQL.sql b/docs/sql/IM-共库运行态缺口修复SQL.sql index 373c1e2..6962c2a 100644 --- a/docs/sql/IM-共库运行态缺口修复SQL.sql +++ b/docs/sql/IM-共库运行态缺口修复SQL.sql @@ -45,3 +45,62 @@ CREATE TABLE IF NOT EXISTS im_login_lockouts ( INDEX idx_im_login_lockouts_user (user_id), INDEX idx_im_login_lockouts_locked (locked_until) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- HTTP write idempotency must be tenant scoped in the shared meeting_im DB. +CREATE TABLE IF NOT EXISTS im_http_idempotency ( + 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, + response_body MEDIUMBLOB, + 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 (id), + 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; + +ALTER TABLE im_http_idempotency + ADD COLUMN tenant_id BIGINT NOT NULL DEFAULT 0 AFTER id; +ALTER TABLE im_http_idempotency + DROP INDEX uk_im_http_idempotency_scope; +ALTER TABLE im_http_idempotency + ADD UNIQUE INDEX uk_im_http_idempotency_scope + (tenant_id, user_id, method, route, idempotency_key); + +-- User identity indexes must be unique inside one tenant, not globally. +ALTER TABLE im_users + DROP INDEX uk_im_users_account_no, + DROP INDEX users_phone, + DROP INDEX users_username, + ADD UNIQUE INDEX uk_im_users_tenant_account_no (tenant_id, account_no), + ADD UNIQUE INDEX uk_im_users_tenant_phone (tenant_id, phone), + ADD UNIQUE INDEX uk_im_users_tenant_username (tenant_id, username); + +ALTER TABLE im_auth + DROP INDEX auth_userid, + ADD UNIQUE INDEX auth_tenant_userid (tenant_id, userid); + +-- Default app release rows used by /v1/im/app/releases/latest. +INSERT INTO hotgo.im_app_releases ( + platform, channel, latest_version, min_supported_version, update_type, + release_notes, remind_after_seconds +) VALUES + ('ios', 'stable', '1.0.0', '1.0.0', 'none', '当前已是最新版本。', 0), + ('android', 'stable', '1.0.0', '1.0.0', 'none', '当前已是最新版本。', 0), + ('web', 'stable', '1.0.0', '1.0.0', 'none', '当前已是最新版本。', 0) +ON DUPLICATE KEY UPDATE + latest_version = VALUES(latest_version), + min_supported_version = VALUES(min_supported_version), + update_type = VALUES(update_type), + release_notes = VALUES(release_notes), + remind_after_seconds = VALUES(remind_after_seconds), + updated_at = CURRENT_TIMESTAMP(3); diff --git a/server/db/mysql/infrastructure_schema.go b/server/db/mysql/infrastructure_schema.go index 362c7e2..e4389dc 100644 --- a/server/db/mysql/infrastructure_schema.go +++ b/server/db/mysql/infrastructure_schema.go @@ -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 } diff --git a/server/db/mysql/infrastructure_schema_test.go b/server/db/mysql/infrastructure_schema_test.go index 45eadaa..e92786d 100644 --- a/server/db/mysql/infrastructure_schema_test.go +++ b/server/db/mysql/infrastructure_schema_test.go @@ -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) } diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index a1f906e..e2936a4 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -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 } } diff --git a/server/db/mysql/tests/mysql_test.go b/server/db/mysql/tests/mysql_test.go index 60f8879..ae0b98f 100644 --- a/server/db/mysql/tests/mysql_test.go +++ b/server/db/mysql/tests/mysql_test.go @@ -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) diff --git a/server/db/mysql/user_identity_schema.go b/server/db/mysql/user_identity_schema.go index d007b59..c1b7654 100644 --- a/server/db/mysql/user_identity_schema.go +++ b/server/db/mysql/user_identity_schema.go @@ -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 } diff --git a/server/im_http.go b/server/im_http.go index d9efad6..034cf2d 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -238,6 +238,8 @@ func serveIMHTTPIdempotent(wrt http.ResponseWriter, req *http.Request, next http } existing, acquired, err := store.HTTPInfrastructure.BeginIdempotency(principal.TenantID, record) if err != nil { + logs.Warn.Printf("im http idempotency begin failed request_id=%s tenant=%d user=%s method=%s route=%s err=%v", + req.Header.Get("X-Request-ID"), principal.TenantID, principal.Uid.UserId(), req.Method, req.URL.Path, err) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return } @@ -262,12 +264,20 @@ func serveIMHTTPIdempotent(wrt http.ResponseWriter, req *http.Request, next http record.ResponseStatus = buffered.status record.ResponseBody = append([]byte(nil), buffered.body.Bytes()...) if buffered.status >= http.StatusInternalServerError { - _ = store.HTTPInfrastructure.AbortIdempotency(principal.TenantID, record) + if abortErr := store.HTTPInfrastructure.AbortIdempotency(principal.TenantID, record); abortErr != nil { + logs.Warn.Printf("im http idempotency abort failed request_id=%s tenant=%d user=%s method=%s route=%s err=%v", + req.Header.Get("X-Request-ID"), principal.TenantID, principal.Uid.UserId(), req.Method, req.URL.Path, abortErr) + } buffered.flushTo(wrt) return } if err = store.HTTPInfrastructure.CompleteIdempotency(principal.TenantID, record); err != nil { - _ = store.HTTPInfrastructure.AbortIdempotency(principal.TenantID, record) + logs.Warn.Printf("im http idempotency complete failed request_id=%s tenant=%d user=%s method=%s route=%s status=%d err=%v", + req.Header.Get("X-Request-ID"), principal.TenantID, principal.Uid.UserId(), req.Method, req.URL.Path, buffered.status, err) + if abortErr := store.HTTPInfrastructure.AbortIdempotency(principal.TenantID, record); abortErr != nil { + logs.Warn.Printf("im http idempotency abort after complete failed request_id=%s tenant=%d user=%s method=%s route=%s err=%v", + req.Header.Get("X-Request-ID"), principal.TenantID, principal.Uid.UserId(), req.Method, req.URL.Path, abortErr) + } writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return } @@ -2063,8 +2073,21 @@ func serveIMAppReleaseLatest(wrt http.ResponseWriter, req *http.Request) { } release, err := store.AppReleases.Latest(platform, channel) if err != nil { - writeIMRelationError(wrt, req, err) - return + if platform == "web" { + logs.Warn.Println("im app release latest: using web fallback", err) + release = &types.AppRelease{ + Platform: platform, + Channel: channel, + LatestVersion: currentVersion, + MinSupportedVersion: currentVersion, + UpdateType: "none", + ReleaseNotes: "当前已是最新版本。", + PublishedAt: time.Now().UTC(), + } + } else { + writeIMRelationError(wrt, req, err) + return + } } latestParts, latestErr := parseNumericAppVersion(release.LatestVersion) minimumParts, minimumErr := parseNumericAppVersion(release.MinSupportedVersion) diff --git a/server/im_http_test.go b/server/im_http_test.go index 53e7fd1..ae7a9dd 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -265,6 +265,7 @@ type testProfilesStore struct { type testAppReleaseStore struct { release *types.AppRelease + err error } type testDevicePushTokenStore struct { @@ -756,6 +757,9 @@ func (s *testProfilesStore) UpdatePrivacy(_ types.TenantID, uid types.Uid, patch } func (s *testAppReleaseStore) Latest(platform, channel string) (*types.AppRelease, error) { + if s.err != nil { + return nil, s.err + } if s.release == nil { now := time.Now().UTC().Round(time.Millisecond) s.release = &types.AppRelease{ @@ -2742,6 +2746,26 @@ func TestServeIMAppReleaseLatestSupportsWeb(t *testing.T) { } } +func TestServeIMAppReleaseLatestFallsBackForWebConfigError(t *testing.T) { + oldReleases := store.AppReleases + store.AppReleases = &testAppReleaseStore{err: types.ErrUnsupported} + defer func() { + store.AppReleases = oldReleases + }() + + req := httptest.NewRequest(http.MethodGet, "/v1/im/app/releases/latest?platform=web¤t_version=1.0.0", nil) + rec := httptest.NewRecorder() + serveIMAppReleaseLatest(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) + if data["platform"] != "web" || data["latest_version"] != "1.0.0" || data["update_type"] != "none" || data["has_update"] != false { + t.Fatalf("unexpected fallback release payload: %#v", data) + } +} + func TestServeIMAppReleaseLatestForcesUnsupportedVersion(t *testing.T) { oldReleases := store.AppReleases store.AppReleases = &testAppReleaseStore{release: &types.AppRelease{