Austin
b2e8b64c0b
重构MySQL迁移逻辑，新增acquireMigrationLock适配Percona XtraDB Cluster的pxc_strict_mode限制，将原有的GET_LOCK迁移锁替换为更兼容的方案，同时为各迁移流程添加快速检查路径减少锁竞争
706 lines
25 KiB
Go
706 lines
25 KiB
Go
//go:build mysql
|
|
// +build mysql
|
|
|
|
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"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()
|
|
|
|
// 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)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if unlock != nil {
|
|
defer unlock()
|
|
}
|
|
|
|
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 '',
|
|
friend_add_enabled TINYINT NOT NULL DEFAULT 1,
|
|
friend_add_disabled_reason VARCHAR(255) NOT NULL DEFAULT '',
|
|
friend_add_policy_version INT NOT NULL DEFAULT 1,
|
|
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 exists, err := identityColumnExists(ctx, conn, "im_users", "username"); err != nil {
|
|
return err
|
|
} else if !exists {
|
|
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_users ADD COLUMN username VARCHAR(32) NULL AFTER id"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err = ensureUserAccountNoSchema(ctx, conn); 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 exists, err := identityColumnExists(ctx, conn, "im_user_profiles", "friend_add_enabled"); err != nil {
|
|
return err
|
|
} else if !exists {
|
|
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_user_profiles ADD COLUMN friend_add_enabled TINYINT NOT NULL DEFAULT 1 AFTER public_account"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if exists, err := identityColumnExists(ctx, conn, "im_user_profiles", "friend_add_disabled_reason"); err != nil {
|
|
return err
|
|
} else if !exists {
|
|
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_user_profiles ADD COLUMN friend_add_disabled_reason VARCHAR(255) NOT NULL DEFAULT '' AFTER friend_add_enabled"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if exists, err := identityColumnExists(ctx, conn, "im_user_profiles", "friend_add_policy_version"); err != nil {
|
|
return err
|
|
} else if !exists {
|
|
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_user_profiles ADD COLUMN friend_add_policy_version INT NOT NULL DEFAULT 1 AFTER friend_add_disabled_reason"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
if err = backfillUserPhones(ctx, conn, usersScoped); err != nil {
|
|
return err
|
|
}
|
|
if err = backfillUsernames(ctx, conn, usersScoped); err != nil {
|
|
return err
|
|
}
|
|
if profilesExist {
|
|
if err = backfillProfileEmails(ctx, conn, usersScoped); err != nil {
|
|
return err
|
|
}
|
|
if err = backfillProfileDisplay(ctx, conn, usersScoped); err != nil {
|
|
return err
|
|
}
|
|
if err = backfillUserPublicFromProfile(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)
|
|
}
|
|
}
|
|
usernameIndex := "users_username"
|
|
usernameColumns := "username"
|
|
if usersScoped {
|
|
usernameIndex = "uk_im_users_tenant_username"
|
|
usernameColumns = "tenant_id,username"
|
|
}
|
|
if exists, err := identityIndexExists(ctx, conn, "im_users", usernameIndex); err != nil {
|
|
return err
|
|
} else if !exists {
|
|
if _, err = conn.ExecContext(ctx, "CREATE UNIQUE INDEX "+quoteIdent(usernameIndex)+" ON im_users("+usernameColumns+")"); err != nil {
|
|
return fmt.Errorf("create unique username index: %w", err)
|
|
}
|
|
}
|
|
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_users MODIFY COLUMN username VARCHAR(32) NOT NULL"); err != nil {
|
|
return err
|
|
}
|
|
if err = ensureUserAccountSequence(ctx, conn); err != nil {
|
|
return err
|
|
}
|
|
if err = migrateBasicAuthSchema(ctx, conn, usersScoped); err != nil {
|
|
return 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 ensureUserAccountSequence(ctx context.Context, conn *sqlx.Conn) error {
|
|
exists, err := identityColumnExists(ctx, conn, "im_users", "account_no")
|
|
if err != nil || !exists {
|
|
return err
|
|
}
|
|
if _, err = conn.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS im_user_account_sequences(
|
|
id TINYINT UNSIGNED NOT NULL,next_value INT UNSIGNED NOT NULL,
|
|
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
|
|
PRIMARY KEY(id)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`); err != nil {
|
|
return err
|
|
}
|
|
_, err = conn.ExecContext(ctx, `INSERT INTO im_user_account_sequences(id,next_value)
|
|
SELECT 1,GREATEST(COALESCE(MAX(account_no),999),999) FROM im_users
|
|
ON DUPLICATE KEY UPDATE next_value=GREATEST(next_value,VALUES(next_value))`)
|
|
return err
|
|
}
|
|
|
|
func backfillUsernames(ctx context.Context, conn *sqlx.Conn, scoped bool) error {
|
|
prefix := "0 AS tenant_id,u.id AS user_id,"
|
|
join := "a.userid=u.id"
|
|
orderBy := "u.id"
|
|
if scoped {
|
|
prefix = "u.tenant_id AS tenant_id,u.id AS user_id,"
|
|
join = "a.tenant_id=u.tenant_id AND a.userid=u.id"
|
|
orderBy = "u.tenant_id,u.id"
|
|
}
|
|
authExists, err := identityTableExists(ctx, conn, "im_auth")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
selectAuth := "NULL"
|
|
authJoin := ""
|
|
if authExists {
|
|
hasUname, err := identityColumnExists(ctx, conn, "im_auth", "uname")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
hasScheme, err := identityColumnExists(ctx, conn, "im_auth", "scheme")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if hasUname {
|
|
selectAuth = "CASE WHEN a.uname LIKE 'basic:%' THEN SUBSTRING(a.uname,7) ELSE a.uname END"
|
|
authJoin = " LEFT JOIN im_auth AS a ON " + join
|
|
if hasScheme {
|
|
authJoin += " AND a.scheme='basic'"
|
|
}
|
|
}
|
|
}
|
|
var rows []identityValue
|
|
query := "SELECT " + prefix + "COALESCE(NULLIF(u.username,''),NULLIF(" + selectAuth + ",''),NULLIF(u.phone,''),CONCAT('reserved_',u.id)) AS value FROM im_users AS u" + authJoin + " WHERE u.username IS NULL OR u.username='' ORDER BY " + orderBy
|
|
if err = conn.SelectContext(ctx, &rows, query); err != nil {
|
|
return err
|
|
}
|
|
seen := make(map[string]bool, len(rows))
|
|
for _, row := range rows {
|
|
value := strings.TrimSpace(row.Value)
|
|
if strings.HasPrefix(value, "+") || (len(value) == 11 && asciiDigits(value)) {
|
|
if canonical, normalizeErr := normalizeUserPhone(value); normalizeErr == nil {
|
|
value = canonical
|
|
}
|
|
}
|
|
if value == "" || len(value) > 32 || seen[identityUniqueKey(row.TenantID, value, scoped)] {
|
|
value = fmt.Sprintf("reserved_%d", row.UserID)
|
|
}
|
|
seen[identityUniqueKey(row.TenantID, value, scoped)] = true
|
|
update := "UPDATE im_users SET username=? WHERE id=?"
|
|
args := []any{value, row.UserID}
|
|
if scoped {
|
|
update = "UPDATE im_users SET username=? WHERE tenant_id=? AND id=?"
|
|
args = []any{value, row.TenantID, row.UserID}
|
|
}
|
|
if _, err = conn.ExecContext(ctx, update, args...); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func migrateBasicAuthSchema(ctx context.Context, conn *sqlx.Conn, scoped bool) error {
|
|
exists, err := identityTableExists(ctx, conn, "im_auth")
|
|
if err != nil || !exists {
|
|
return err
|
|
}
|
|
if hasScheme, err := identityColumnExists(ctx, conn, "im_auth", "scheme"); err != nil {
|
|
return err
|
|
} else if hasScheme {
|
|
var unsupported int
|
|
if err = conn.QueryRowContext(ctx, "SELECT COUNT(*) FROM im_auth WHERE scheme<>'basic'").Scan(&unsupported); err != nil {
|
|
return err
|
|
}
|
|
if unsupported > 0 {
|
|
return fmt.Errorf("mysql: im_auth contains %d non-basic records", unsupported)
|
|
}
|
|
}
|
|
legacyComposite, err := identityIndexExists(ctx, conn, "im_auth", "auth_userid_scheme")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if legacyComposite {
|
|
// The legacy composite index may currently support the userid foreign key.
|
|
// Give the FK a replacement before dropping that index.
|
|
if exists, err = identityIndexExists(ctx, conn, "im_auth", "auth_userid_fk"); err != nil {
|
|
return err
|
|
} else if !exists {
|
|
if _, err = conn.ExecContext(ctx, "CREATE INDEX auth_userid_fk ON im_auth(userid)"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
for _, index := range []string{"auth_uname", "auth_userid_scheme"} {
|
|
if exists, err = identityIndexExists(ctx, conn, "im_auth", index); err != nil {
|
|
return err
|
|
} else if exists {
|
|
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_auth DROP INDEX "+quoteIdent(index)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
for _, column := range []string{"uname", "scheme"} {
|
|
if exists, err = identityColumnExists(ctx, conn, "im_auth", column); err != nil {
|
|
return err
|
|
} else if exists {
|
|
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_auth DROP COLUMN "+quoteIdent(column)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
indexName := "auth_userid"
|
|
columns := "userid"
|
|
if scoped {
|
|
indexName = "auth_tenant_userid"
|
|
columns = "tenant_id,userid"
|
|
}
|
|
if exists, err = identityIndexExists(ctx, conn, "im_auth", indexName); err != nil {
|
|
return err
|
|
} else if !exists {
|
|
if _, err = conn.ExecContext(ctx, "CREATE UNIQUE INDEX "+quoteIdent(indexName)+" ON im_auth("+columns+")"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if !scoped {
|
|
if exists, err = identityIndexExists(ctx, conn, "im_auth", "auth_userid_fk"); err != nil {
|
|
return err
|
|
} else if exists {
|
|
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_auth DROP INDEX auth_userid_fk"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func asciiDigits(value string) bool {
|
|
if value == "" {
|
|
return false
|
|
}
|
|
for i := range value {
|
|
if value[i] < '0' || value[i] > '9' {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
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 backfillProfileDisplay(ctx context.Context, conn *sqlx.Conn, usersScoped bool) error {
|
|
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")
|
|
}
|
|
nicknameExpr := `COALESCE(
|
|
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.display_name')), ''),
|
|
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.nickname')), ''),
|
|
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.fn')), ''),
|
|
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.name')), '')
|
|
)`
|
|
avatarExpr := `COALESCE(
|
|
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.avatar_file_id')), '')
|
|
)`
|
|
if usersScoped {
|
|
if _, err = conn.ExecContext(ctx, `INSERT IGNORE INTO im_user_profiles(id,tenant_id,user_id,nickname,avatar_file_id,public_account)
|
|
SELECT u.id,u.tenant_id,u.id,COALESCE(`+nicknameExpr+`,''),COALESCE(`+avatarExpr+`,''),''
|
|
FROM im_users AS u
|
|
WHERE `+nicknameExpr+` IS NOT NULL OR `+avatarExpr+` IS NOT NULL`); err != nil {
|
|
return err
|
|
}
|
|
_, err = conn.ExecContext(ctx, `UPDATE im_user_profiles AS p
|
|
JOIN im_users AS u ON u.tenant_id=p.tenant_id AND u.id=p.user_id
|
|
SET p.nickname=IF(p.nickname='' AND `+nicknameExpr+` IS NOT NULL, `+nicknameExpr+`, p.nickname),
|
|
p.avatar_file_id=IF(p.avatar_file_id='' AND `+avatarExpr+` IS NOT NULL, `+avatarExpr+`, p.avatar_file_id)
|
|
WHERE (p.nickname='' AND `+nicknameExpr+` IS NOT NULL) OR (p.avatar_file_id='' AND `+avatarExpr+` IS NOT NULL)`)
|
|
return err
|
|
}
|
|
if _, err = conn.ExecContext(ctx, `INSERT IGNORE INTO im_user_profiles(id,user_id,nickname,avatar_file_id,public_account)
|
|
SELECT u.id,u.id,COALESCE(`+nicknameExpr+`,''),COALESCE(`+avatarExpr+`,''),''
|
|
FROM im_users AS u
|
|
WHERE `+nicknameExpr+` IS NOT NULL OR `+avatarExpr+` IS NOT NULL`); err != nil {
|
|
return err
|
|
}
|
|
_, err = conn.ExecContext(ctx, `UPDATE im_user_profiles AS p
|
|
JOIN im_users AS u ON u.id=p.user_id
|
|
SET p.nickname=IF(p.nickname='' AND `+nicknameExpr+` IS NOT NULL, `+nicknameExpr+`, p.nickname),
|
|
p.avatar_file_id=IF(p.avatar_file_id='' AND `+avatarExpr+` IS NOT NULL, `+avatarExpr+`, p.avatar_file_id)
|
|
WHERE (p.nickname='' AND `+nicknameExpr+` IS NOT NULL) OR (p.avatar_file_id='' AND `+avatarExpr+` IS NOT NULL)`)
|
|
return err
|
|
}
|
|
|
|
func backfillUserPublicFromProfile(ctx context.Context, conn *sqlx.Conn, usersScoped bool) error {
|
|
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")
|
|
}
|
|
publicDoc := "COALESCE(NULLIF(u.public,''), JSON_OBJECT())"
|
|
nicknamePatch := `u.public=JSON_SET(` + publicDoc + `,
|
|
'$.display_name',p.nickname,'$.nickname',p.nickname,'$.fn',p.nickname,'$.name',p.nickname)`
|
|
avatarPatch := `u.public=JSON_SET(` + publicDoc + `,'$.avatar_file_id',p.avatar_file_id)`
|
|
accountPatch := `u.public=JSON_SET(` + publicDoc + `,'$.public_account',p.public_account)`
|
|
join := "JOIN im_users AS u ON u.id=p.user_id"
|
|
if usersScoped {
|
|
join = "JOIN im_users AS u ON u.tenant_id=p.tenant_id AND u.id=p.user_id"
|
|
}
|
|
for _, query := range []string{
|
|
`UPDATE im_user_profiles AS p ` + join + `
|
|
SET ` + nicknamePatch + `
|
|
WHERE p.nickname<>'' AND (
|
|
COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.display_name')),'')<>p.nickname OR
|
|
COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.nickname')),'')<>p.nickname OR
|
|
COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.fn')),'')<>p.nickname OR
|
|
COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.name')),'')<>p.nickname)`,
|
|
`UPDATE im_user_profiles AS p ` + join + `
|
|
SET ` + avatarPatch + `
|
|
WHERE p.avatar_file_id<>'' AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.avatar_file_id')),'')<>p.avatar_file_id`,
|
|
`UPDATE im_user_profiles AS p ` + join + `
|
|
SET ` + accountPatch + `
|
|
WHERE p.public_account<>'' AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.public_account')),'')<>p.public_account`,
|
|
} {
|
|
if _, err = conn.ExecContext(ctx, query); 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 {
|
|
hasUname, err := identityColumnExists(ctx, conn, "im_auth", "uname")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
hasScheme, err := identityColumnExists(ctx, conn, "im_auth", "scheme")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !hasUname || !hasScheme {
|
|
return out, nil
|
|
}
|
|
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 ensureUserAccountNoSchema(ctx context.Context, conn *sqlx.Conn) error {
|
|
if exists, err := identityColumnExists(ctx, conn, "im_users", "account_no"); err != nil {
|
|
return err
|
|
} else if !exists {
|
|
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_users ADD COLUMN account_no INT UNSIGNED NOT NULL DEFAULT 0 AFTER id"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
var start int64 = 999
|
|
if err := conn.QueryRowContext(ctx, "SELECT GREATEST(COALESCE(MAX(account_no), 999), 999) FROM im_users WHERE account_no>0").Scan(&start); err != nil {
|
|
return err
|
|
}
|
|
if _, err := conn.ExecContext(ctx, `
|
|
UPDATE im_users AS u
|
|
JOIN (
|
|
SELECT ordered.id, (@im_user_account_no := @im_user_account_no + 1) AS next_no
|
|
FROM (SELECT id FROM im_users WHERE account_no=0 ORDER BY createdat, id) AS ordered
|
|
CROSS JOIN (SELECT @im_user_account_no := ?) AS vars
|
|
) AS seq ON seq.id=u.id
|
|
SET u.account_no=seq.next_no
|
|
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 {
|
|
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 {
|
|
return fmt.Errorf("create unique user account number index: %w", err)
|
|
}
|
|
}
|
|
return 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
|
|
}
|