//go:build mysql // +build mysql package mysql import ( "context" "database/sql" "errors" "fmt" "strings" "github.com/jmoiron/sqlx" "github.com/tinode/chat/server/store" ) const userIdentityMigrationLock = "CONCAT('im_user_identity:', LEFT(SHA2(DATABASE(), 256), 40))" type identityValue struct { TenantID int64 `db:"tenant_id"` UserID int64 `db:"user_id"` Value string `db:"value"` } // ensureUserIdentitySchema moves user phone/email out of the legacy tag store. // Topic tags are intentionally unaffected. func ensureUserIdentitySchema(db *sqlx.DB) error { ctx := context.Background() conn, err := db.Connx(ctx) if err != nil { return err } defer conn.Close() var locked sql.NullInt64 if err = conn.QueryRowContext(ctx, "SELECT GET_LOCK("+userIdentityMigrationLock+", 60)").Scan(&locked); err != nil { return err } if !locked.Valid || locked.Int64 != 1 { return errors.New("mysql: timed out waiting for user identity migration lock") } defer conn.ExecContext(ctx, "SELECT RELEASE_LOCK("+userIdentityMigrationLock+")") usersExist, err := identityTableExists(ctx, conn, "im_users") if err != nil || !usersExist { return err } usersScoped, err := identityColumnExists(ctx, conn, "im_users", "tenant_id") if err != nil { return err } profilesExist, err := identityTableExists(ctx, conn, "im_user_profiles") if err != nil { return err } if !profilesExist { profileScope := "" profileUniqueKey := "user_id" if usersScoped { profileScope = "tenant_id BIGINT NOT NULL," profileUniqueKey = "tenant_id,user_id" } if _, err = conn.ExecContext(ctx, `CREATE TABLE im_user_profiles( id BIGINT NOT NULL,`+profileScope+`user_id BIGINT NOT NULL, avatar_file_id VARCHAR(64) NOT NULL DEFAULT '',nickname VARCHAR(64) NOT NULL DEFAULT '', birth_date DATE,signature VARCHAR(256) NOT NULL DEFAULT '',gender VARCHAR(16) NOT NULL DEFAULT '', region VARCHAR(128) NOT NULL DEFAULT '',email VARCHAR(128) NOT NULL DEFAULT '', public_account VARCHAR(64) NOT NULL DEFAULT '',version INT NOT NULL DEFAULT 1, 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_user_profiles_user(`+profileUniqueKey+`))`); err != nil { return err } profilesExist = true } if exists, err := identityColumnExists(ctx, conn, "im_users", "phone"); err != nil { return err } else if !exists { if _, err = conn.ExecContext(ctx, "ALTER TABLE im_users ADD COLUMN phone VARCHAR(32) NULL AFTER useragent"); err != nil { return err } } if profilesExist { if exists, err := identityColumnExists(ctx, conn, "im_user_profiles", "email"); err != nil { return err } else if !exists { if _, err = conn.ExecContext(ctx, "ALTER TABLE im_user_profiles ADD COLUMN email VARCHAR(128) NOT NULL DEFAULT '' AFTER region"); err != nil { return err } } } if err = backfillUserPhones(ctx, conn, usersScoped); err != nil { return err } if profilesExist { if err = backfillProfileEmails(ctx, conn, usersScoped); err != nil { return err } } indexName := "users_phone" if usersScoped { indexName = "uk_im_users_tenant_phone" } if exists, err := identityIndexExists(ctx, conn, "im_users", indexName); err != nil { return err } else if !exists { columns := "phone" if usersScoped { columns = "tenant_id,phone" } if _, err = conn.ExecContext(ctx, "CREATE UNIQUE INDEX "+quoteIdent(indexName)+" ON im_users("+columns+")"); err != nil { return fmt.Errorf("create unique phone index: %w", err) } } if exists, err := identityTableExists(ctx, conn, "im_usertags"); err != nil { return err } else if exists { if _, err = conn.ExecContext(ctx, "DROP TABLE im_usertags"); err != nil { return err } } if exists, err := identityColumnExists(ctx, conn, "im_users", "tags"); err != nil { return err } else if exists { if _, err = conn.ExecContext(ctx, "ALTER TABLE im_users DROP COLUMN tags"); err != nil { return err } } return nil } func backfillUserPhones(ctx context.Context, conn *sqlx.Conn, scoped bool) error { candidates, err := identityCandidates(ctx, conn, scoped, "tel") if err != nil { return err } seen := make(map[string]bool) var existing []identityValue prefix := "0 AS tenant_id," if scoped { prefix = "tenant_id," } orderBy := "id" if scoped { orderBy = "tenant_id,id" } if err = conn.SelectContext(ctx, &existing, "SELECT "+prefix+"id AS user_id,phone AS value FROM im_users WHERE phone IS NOT NULL AND phone<>'' ORDER BY "+orderBy); err != nil { return err } for _, item := range existing { canonical, err := normalizeUserPhone(item.Value) if err != nil { return fmt.Errorf("normalize existing phone for user %d: %w", item.UserID, err) } if canonical != item.Value { query := "UPDATE im_users SET phone=? WHERE id=?" args := []any{canonical, item.UserID} if scoped { query = "UPDATE im_users SET phone=? WHERE tenant_id=? AND id=?" args = []any{canonical, item.TenantID, item.UserID} } if _, err = conn.ExecContext(ctx, query, args...); err != nil { return err } } seen[identityUniqueKey(item.TenantID, canonical, scoped)] = true } updatedUsers := make(map[string]bool) for _, item := range candidates { item.Value, err = normalizeUserPhone(item.Value) if err != nil { return fmt.Errorf("normalize legacy phone for user %d: %w", item.UserID, err) } userKey := identityUserKey(item.TenantID, item.UserID, scoped) valueKey := identityUniqueKey(item.TenantID, item.Value, scoped) if item.Value == "" || updatedUsers[userKey] || seen[valueKey] { continue } query := "UPDATE im_users SET phone=? WHERE id=? AND (phone IS NULL OR phone='')" args := []any{item.Value, item.UserID} if scoped { query = "UPDATE im_users SET phone=? WHERE tenant_id=? AND id=? AND (phone IS NULL OR phone='')" args = []any{item.Value, item.TenantID, item.UserID} } result, err := conn.ExecContext(ctx, query, args...) if err != nil { return err } if affected, _ := result.RowsAffected(); affected > 0 { updatedUsers[userKey] = true seen[valueKey] = true } } return nil } func backfillProfileEmails(ctx context.Context, conn *sqlx.Conn, usersScoped bool) error { candidates, err := identityCandidates(ctx, conn, usersScoped, "email") if err != nil { return err } profilesScoped, err := identityColumnExists(ctx, conn, "im_user_profiles", "tenant_id") if err != nil { return err } if profilesScoped != usersScoped { return errors.New("mysql: im_users and im_user_profiles tenant scope mismatch") } processed := make(map[string]bool) for _, item := range candidates { item.Value = strings.TrimSpace(strings.TrimPrefix(item.Value, "email:")) key := identityUserKey(item.TenantID, item.UserID, usersScoped) if item.Value == "" || processed[key] { continue } processed[key] = true id, err := nextSnowflakeID() if err != nil { return err } account := store.EncodeUid(item.UserID).UserId() if usersScoped { _, err = conn.ExecContext(ctx, `INSERT INTO im_user_profiles(id,tenant_id,user_id,email,public_account) VALUES(?,?,?,?,?) ON DUPLICATE KEY UPDATE email=IF(email='',VALUES(email),email)`, id, item.TenantID, item.UserID, item.Value, account) } else { _, err = conn.ExecContext(ctx, `INSERT INTO im_user_profiles(id,user_id,email,public_account) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE email=IF(email='',VALUES(email),email)`, id, item.UserID, item.Value, account) } if err != nil { return err } } return nil } func identityCandidates(ctx context.Context, conn *sqlx.Conn, scoped bool, method string) ([]identityValue, error) { var out []identityValue appendQuery := func(query string, args ...any) error { var rows []identityValue if err := conn.SelectContext(ctx, &rows, query, args...); err != nil { return err } out = append(out, rows...) return nil } prefix := "0 AS tenant_id,u.id AS user_id," credentialJoin := "c.userid=u.id" tagJoin := "ut.userid=u.id" if scoped { prefix = "u.tenant_id AS tenant_id,u.id AS user_id," credentialJoin = "c.tenant_id=u.tenant_id AND c.userid=u.id" tagJoin = "ut.tenant_id=u.tenant_id AND ut.userid=u.id" } if exists, err := identityTableExists(ctx, conn, "im_credentials"); err != nil { return nil, err } else if exists { if err = appendQuery("SELECT "+prefix+"c.value AS value FROM im_users AS u JOIN im_credentials AS c ON "+credentialJoin+ " WHERE c.method=? AND c.done=1 AND c.deletedat IS NULL ORDER BY u.id,c.updatedat DESC", method); err != nil { return nil, err } } if exists, err := identityTableExists(ctx, conn, "im_usertags"); err != nil { return nil, err } else if exists { if err = appendQuery("SELECT "+prefix+"ut.tag AS value FROM im_users AS u JOIN im_usertags AS ut ON "+tagJoin+ " WHERE ut.tag LIKE ? ORDER BY u.id,ut.id", method+":%"); err != nil { return nil, err } } if method == "tel" { if exists, err := identityTableExists(ctx, conn, "im_auth"); err != nil { return nil, err } else if exists { authJoin := "a.userid=u.id" if scoped { authJoin = "a.tenant_id=u.tenant_id AND a.userid=u.id" } if err = appendQuery("SELECT " + prefix + "a.uname AS value FROM im_users AS u JOIN im_auth AS a ON " + authJoin + " WHERE a.scheme='basic' AND (a.uname LIKE '+%' OR a.uname LIKE 'basic:+%') ORDER BY u.id"); err != nil { return nil, err } } } return out, nil } func identityTableExists(ctx context.Context, conn *sqlx.Conn, table string) (bool, error) { var exists int err := conn.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name=?)`, table).Scan(&exists) return exists == 1, err } func identityColumnExists(ctx context.Context, conn *sqlx.Conn, table, column string) (bool, error) { var exists int err := conn.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? AND column_name=?)`, table, column).Scan(&exists) return exists == 1, err } func identityIndexExists(ctx context.Context, conn *sqlx.Conn, table, index string) (bool, error) { var exists int err := conn.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name=? AND index_name=?)`, table, index).Scan(&exists) return exists == 1, err } func identityUserKey(tenantID, userID int64, scoped bool) string { if scoped { return fmt.Sprintf("%d:%d", tenantID, userID) } return fmt.Sprintf("%d", userID) } func identityUniqueKey(tenantID int64, value string, scoped bool) string { if scoped { return fmt.Sprintf("%d:%s", tenantID, value) } return value }