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

feat: migrate IM identity and snowflake primary keys

Browse Source
This commit is contained in:
richjohnrambo
2026-08-18 01:15:21 +08:00
parent db322d80dd
commit bf562c7d56
27 changed files with 1071 additions and 482 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/adapter.go
+123 -151
View File
@@ -246,6 +246,12 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error {
if err = ensureIMHTTPInfrastructureTables(a.db); err != nil {
return errors.New("mysql adapter failed to ensure HTTP infrastructure schema: " + err.Error())
}
if err = ensureSnowflakePrimaryKeys(a.db); err != nil {
return errors.New("mysql adapter failed to ensure snowflake primary keys: " + err.Error())
}
if err = ensureUserIdentitySchema(a.db); err != nil {
return errors.New("mysql adapter failed to ensure user identity schema: " + err.Error())
}
}
}
return err
@@ -440,6 +446,14 @@ func (a *adapter) TenantAdapterFor(tenantID t.TenantID) (dbadapter.TenantBusines
db.Close()
return nil, nil, err
}
if err = ensureSnowflakePrimaryKeys(db); err != nil {
db.Close()
return nil, nil, err
}
if err = ensureUserIdentitySchema(db); err != nil {
db.Close()
return nil, nil, err
}
tenantAdp := &adapter{
db: db,
dsn: mysqlDefaultString(dsn, a.dsn),
@@ -589,7 +603,13 @@ func (a *adapter) CreateDb(reset bool) error {
if _, err = tx.Exec("DROP DATABASE IF EXISTS " + quoteIdent(businessDB)); err != nil {
return err
}
if _, err = tx.Exec("DROP DATABASE IF EXISTS " + quoteIdent(registryDB)); err != nil {
if _, err = tx.Exec("CREATE DATABASE IF NOT EXISTS " + quoteIdent(registryDB) + " CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci"); err != nil {
return err
}
if _, err = tx.Exec("DROP TABLE IF EXISTS " + quoteIdent(registryDB) + ".im_tenant_config"); err != nil {
return err
}
if _, err = tx.Exec("DROP TABLE IF EXISTS " + quoteIdent(registryDB) + ".im_tenant"); err != nil {
return err
}
}
@@ -616,10 +636,11 @@ func (a *adapter) CreateDb(reset bool) error {
access JSON,
lastseen DATETIME,
useragent VARCHAR(255) DEFAULT '',
phone VARCHAR(32),
public JSON,
trusted JSON,
tags JSON,
PRIMARY KEY(id),
UNIQUE INDEX users_phone(phone),
INDEX users_state_stateat(state, stateat),
INDEX users_lastseen_updatedat(lastseen, updatedat)
)`); err != nil {
@@ -628,12 +649,14 @@ func (a *adapter) CreateDb(reset bool) error {
if _, err = tx.Exec(
`CREATE TABLE im_user_security(
id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
auth_version INT NOT NULL DEFAULT 1,
password_updated_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(user_id)
PRIMARY KEY(id),
UNIQUE INDEX uk_im_user_security_user(user_id)
)`); err != nil {
return err
}
@@ -645,20 +668,6 @@ func (a *adapter) CreateDb(reset bool) error {
return err
}
// Indexed user tags.
if _, err = tx.Exec(
`CREATE TABLE im_usertags(
id INT NOT NULL AUTO_INCREMENT,
userid BIGINT NOT NULL,
tag VARCHAR(96) NOT NULL,
PRIMARY KEY(id),
FOREIGN KEY(userid) REFERENCES im_users(id),
INDEX usertags_tag(tag),
UNIQUE INDEX usertags_userid_tag(userid, tag)
)`); err != nil {
return err
}
// Indexed devices. Normalized into a separate table.
if _, err = tx.Exec(
`CREATE TABLE im_devices(
@@ -919,15 +928,21 @@ func (a *adapter) CreateDb(reset bool) error {
if _, err = tx.Exec(
`CREATE TABLE im_kvmeta(` +
"id BIGINT NOT NULL," +
"`key` VARCHAR(64) NOT NULL," +
"createdat DATETIME(3)," +
"`value` TEXT," +
"PRIMARY KEY(`key`)," +
"PRIMARY KEY(id)," +
"UNIQUE INDEX uk_im_kvmeta_key(`key`)," +
"INDEX kvmeta_createdat_key(createdat, `key`)" +
`)`); err != nil {
return err
}
if _, err = tx.Exec("INSERT INTO im_kvmeta(`key`, `value`) VALUES('version',?)", adpVersion); err != nil {
versionID, err := nextSnowflakeID()
if err != nil {
return err
}
if _, err = tx.Exec("INSERT INTO im_kvmeta(id, `key`, `value`) VALUES(?,'version',?)", versionID, adpVersion); err != nil {
return err
}
@@ -1225,6 +1240,7 @@ func createIMMailboxTables(tx *sql.Tx) error {
func createIMUserProductTables(tx *sql.Tx) error {
statements := []string{
`CREATE TABLE im_user_profiles(
id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
avatar_file_id VARCHAR(64) NOT NULL DEFAULT '',
nickname VARCHAR(64) NOT NULL DEFAULT '',
@@ -1232,13 +1248,16 @@ func createIMUserProductTables(tx *sql.Tx) error {
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(user_id)
PRIMARY KEY(id),
UNIQUE INDEX uk_im_user_profiles_user(user_id)
)`,
`CREATE TABLE im_user_privacy_settings(
id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
add_me_methods JSON,
profile_visibility VARCHAR(32) NOT NULL DEFAULT 'friends',
@@ -1247,7 +1266,8 @@ func createIMUserProductTables(tx *sql.Tx) error {
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(user_id)
PRIMARY KEY(id),
UNIQUE INDEX uk_im_user_privacy_settings_user(user_id)
)`,
`CREATE TABLE im_app_releases(
id BIGINT NOT NULL AUTO_INCREMENT,
@@ -1268,6 +1288,7 @@ func createIMUserProductTables(tx *sql.Tx) error {
UNIQUE INDEX uk_im_app_releases_platform_channel(platform, channel)
)`,
`CREATE TABLE im_device_push_tokens(
id BIGINT NOT NULL,
device_id VARCHAR(128) NOT NULL,
user_id BIGINT NOT NULL,
provider VARCHAR(32) NOT NULL,
@@ -1276,7 +1297,8 @@ func createIMUserProductTables(tx *sql.Tx) error {
platform VARCHAR(32) NOT NULL DEFAULT '',
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(device_id),
PRIMARY KEY(id),
UNIQUE INDEX uk_im_device_push_tokens_device(device_id),
UNIQUE INDEX uk_im_device_push_tokens_provider_token(provider, token),
INDEX idx_im_device_push_tokens_user(user_id, updated_at)
)`,
@@ -1443,6 +1465,7 @@ func createIMRelationTables(exec sqlExecutor) error {
INDEX idx_im_friend_requests_inbox(target_id, status, created_at)
)`,
`CREATE TABLE im_friendships(
id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
peer_id BIGINT NOT NULL,
status VARCHAR(16) NOT NULL,
@@ -1450,15 +1473,18 @@ func createIMRelationTables(exec sqlExecutor) error {
topic CHAR(25) NOT NULL DEFAULT '',
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(user_id, peer_id),
PRIMARY KEY(id),
UNIQUE INDEX uk_im_friendships_pair(user_id, peer_id),
INDEX idx_im_friendships_peer(peer_id, status)
)`,
`CREATE TABLE im_user_blocks(
id BIGINT NOT NULL,
blocker_id BIGINT NOT NULL,
blocked_id BIGINT 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(blocker_id, blocked_id),
PRIMARY KEY(id),
UNIQUE INDEX uk_im_user_blocks_pair(blocker_id, blocked_id),
INDEX idx_im_user_blocks_blocked(blocked_id)
)`,
`CREATE TABLE im_favorites(
@@ -1536,6 +1562,7 @@ func createIMSecurityTables(tx *sql.Tx) error {
}
_, err := tx.Exec(
`CREATE TABLE im_youth_mode_settings(
id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
purpose VARCHAR(64) NOT NULL,
password_hash VARBINARY(255) NOT NULL,
@@ -1547,7 +1574,8 @@ func createIMSecurityTables(tx *sql.Tx) error {
locked_until 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(user_id, purpose)
PRIMARY KEY(id),
UNIQUE INDEX uk_im_youth_mode_settings_user_purpose(user_id, purpose)
)`)
return err
}
@@ -1650,27 +1678,31 @@ func (a *adapter) UserCreate(user *t.User) error {
}()
decoded_uid := store.DecodeUid(user.Uid())
if _, err = tx.Exec("INSERT INTO im_users(id,createdat,updatedat,state,access,public,trusted,tags) VALUES(?,?,?,?,?,?,?,?)",
securityID, err := nextSnowflakeID()
if err != nil {
return err
}
var phone any
if strings.TrimSpace(user.Phone) != "" {
phone = strings.TrimSpace(user.Phone)
}
if _, err = tx.Exec("INSERT INTO im_users(id,createdat,updatedat,state,access,phone,public,trusted) VALUES(?,?,?,?,?,?,?,?)",
decoded_uid,
user.CreatedAt,
user.UpdatedAt,
user.State,
user.Access,
phone,
common.ToJSON(user.Public),
common.ToJSON(user.Trusted),
user.Tags); err != nil {
common.ToJSON(user.Trusted)); err != nil {
return err
}
if _, err = tx.Exec("INSERT INTO im_user_security(user_id,auth_version,created_at,updated_at) VALUES(?,?,?,?)",
decoded_uid, 1, user.CreatedAt, user.UpdatedAt); err != nil {
return err
}
// Save user's tags to a separate table to make user findable.
if err = addTags(tx, "im_usertags", "userid", decoded_uid, user.Tags, false); err != nil {
if _, err = tx.Exec("INSERT INTO im_user_security(id,user_id,auth_version,created_at,updated_at) VALUES(?,?,?,?,?)",
securityID, decoded_uid, 1, user.CreatedAt, user.UpdatedAt); err != nil {
return err
}
user.Tags = nil
return tx.Commit()
}
@@ -1895,7 +1927,8 @@ func (a *adapter) UserGet(uid t.Uid) (*t.User, error) {
defer cancel()
}
var user t.User
err := a.db.GetContext(ctx, &user, "SELECT * FROM im_users WHERE id=? AND state!=?", store.DecodeUid(uid), t.StateDeleted)
err := a.db.GetContext(ctx, &user, `SELECT id,createdat,updatedat,state,stateat,access,lastseen,useragent,
COALESCE(phone,'') AS phone,public,trusted FROM im_users WHERE id=? AND state!=?`, store.DecodeUid(uid), t.StateDeleted)
if err == nil {
user.SetUid(uid)
user.Public = common.FromJSON(user.Public)
@@ -1925,7 +1958,8 @@ func (a *adapter) UserGetAll(ids ...t.Uid) ([]t.User, error) {
if cancel != nil {
defer cancel()
}
q, uids, _ := sqlx.In("SELECT * FROM im_users WHERE id IN (?) AND state!=?", uids, t.StateDeleted)
q, uids, _ := sqlx.In(`SELECT id,createdat,updatedat,state,stateat,access,lastseen,useragent,
COALESCE(phone,'') AS phone,public,trusted FROM im_users WHERE id IN (?) AND state!=?`, uids, t.StateDeleted)
rows, err := a.db.QueryxContext(ctx, a.db.Rebind(q), uids...)
if err != nil {
return nil, err
@@ -2048,10 +2082,6 @@ func (a *adapter) UserDelete(uid t.Uid, hard bool) error {
return err
}
if _, err = tx.Exec("DELETE FROM im_usertags WHERE userid=?", decoded_uid); err != nil {
return err
}
if _, err = tx.Exec("DELETE FROM im_users WHERE id=?", decoded_uid); err != nil {
return err
}
@@ -2150,7 +2180,11 @@ func (a *adapter) UserUpdate(uid t.Uid, update map[string]any) error {
}
}()
update = withoutUserTags(update)
cols, args := common.UpdateByMap(update)
if len(cols) == 0 {
return tx.Commit()
}
decoded_uid := store.DecodeUid(uid)
args = append(args, decoded_uid)
_, err = tx.Exec("UPDATE im_users SET "+strings.Join(cols, ",")+" WHERE id=?", args...)
@@ -2166,76 +2200,12 @@ func (a *adapter) UserUpdate(uid t.Uid, update map[string]any) error {
}
}
// Tags are also stored in a separate table
if tags := common.ExtractTags(update); tags != nil {
// First delete all user tags
_, err = tx.Exec("DELETE FROM im_usertags WHERE userid=?", decoded_uid)
if err != nil {
return err
}
// Now insert new tags
err = addTags(tx, "im_usertags", "userid", decoded_uid, tags, false)
if err != nil {
return err
}
}
return tx.Commit()
}
// UserUpdateTags adds, removes, or resets user's tags.
// UserUpdateTags is retained for adapter compatibility; user tags are disabled.
func (a *adapter) UserUpdateTags(uid t.Uid, add, remove, reset []string) ([]string, error) {
ctx, cancel := a.getContextForTx()
if cancel != nil {
defer cancel()
}
tx, err := a.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer func() {
if err != nil {
tx.Rollback()
}
}()
decoded_uid := store.DecodeUid(uid)
if reset != nil {
// Delete all tags first if resetting.
_, err = tx.Exec("DELETE FROM im_usertags WHERE userid=?", decoded_uid)
if err != nil {
return nil, err
}
add = reset
remove = nil
}
// Now insert new tags. Ignore duplicates if resetting.
err = addTags(tx, "im_usertags", "userid", decoded_uid, add, reset == nil)
if err != nil {
return nil, err
}
// Delete tags.
err = removeTags(tx, "im_usertags", "userid", decoded_uid, remove)
if err != nil {
return nil, err
}
var allTags []string
err = tx.Select(&allTags, "SELECT tag FROM im_usertags WHERE userid=?", decoded_uid)
if err != nil {
return nil, err
}
_, err = tx.Exec("UPDATE im_users SET tags=? WHERE id=?", t.StringSlice(allTags), decoded_uid)
if err != nil {
return nil, err
}
return allTags, tx.Commit()
return nil, nil
}
// UserGetByCred returns user ID for the given validated credential.
@@ -2245,7 +2215,12 @@ func (a *adapter) UserGetByCred(method, value string) (t.Uid, error) {
defer cancel()
}
var decoded_uid int64
err := a.db.GetContext(ctx, &decoded_uid, "SELECT userid FROM im_credentials WHERE synthetic=?", method+":"+value)
var err error
if method == "tel" {
err = a.db.GetContext(ctx, &decoded_uid, "SELECT id FROM im_users WHERE phone=? AND state!=?", value, t.StateDeleted)
} else {
err = a.db.GetContext(ctx, &decoded_uid, "SELECT userid FROM im_credentials WHERE synthetic=?", method+":"+value)
}
if err == nil {
return store.EncodeUid(decoded_uid), nil
}
@@ -2266,6 +2241,22 @@ func (a *adapter) TenantUserGetByCred(tenantID t.TenantID, method, value string)
if tenantAdp != a {
return tenantAdp.UserGetByCred(method, value)
}
if !a.tenantScopedDB && method == "tel" {
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
var decodedUID int64
err = a.db.GetContext(ctx, &decodedUID,
"SELECT id FROM im_users WHERE tenant_id=? AND phone=? AND state!=?", tenantID, value, t.StateDeleted)
if err == sql.ErrNoRows {
return t.ZeroUid, nil
}
if err != nil {
return t.ZeroUid, err
}
return store.EncodeUid(decodedUID), nil
}
return a.UserGetByCred(method, value)
}
@@ -3329,17 +3320,21 @@ func (a *adapter) SubsDelForUser(user t.Uid, hard bool) error {
return tx.Commit()
}
// Find returns a list of users or group topics who match given tags, such as "email:jdoe@example.com" or "tel:+18003287448".
// Find returns group topics which match the requested discovery tags.
func (a *adapter) Find(caller, promoPrefix string, req [][]string, opt []string, activeOnly bool) ([]t.Subscription, error) {
index := make(map[string]struct{})
allReq := t.FlattenDoubleSlice(req)
allTags := append(append([]string{}, allReq...), opt...)
if len(allTags) == 0 {
return nil, nil
}
var args []any
stateConstraint := ""
if activeOnly {
args = append(args, t.StateOK)
stateConstraint = "u.state=? AND "
stateConstraint = "t.state=? AND "
}
index := make(map[string]struct{})
allReq := t.FlattenDoubleSlice(req)
for _, tag := range append(allReq, opt...) {
for _, tag := range allTags {
args = append(args, tag)
index[tag] = struct{}{}
}
@@ -3352,29 +3347,9 @@ func (a *adapter) Find(caller, promoPrefix string, req [][]string, opt []string,
matcher = "COUNT(*)"
}
query := "SELECT u.id,u.createdat,u.updatedat,0,u.access,0 AS subcnt,u.public,u.trusted,u.tags," + matcher + " AS matches " +
"FROM im_users AS u JOIN im_usertags AS tg ON tg.userid=u.id " +
"WHERE " + stateConstraint + "tg.tag IN (?" + strings.Repeat(",?", len(allReq)+len(opt)-1) + ") " +
"GROUP BY u.id,u.createdat,u.updatedat,u.access,u.public,u.trusted,u.tags "
if len(allReq) > 0 {
q, a := common.DisjunctionSql(req, "tg.tag")
query += q
args = append(args, a...)
}
query += "UNION ALL "
if activeOnly {
args = append(args, t.StateOK)
stateConstraint = "t.state=? AND "
}
for _, tag := range append(allReq, opt...) {
args = append(args, tag)
}
query += "SELECT t.name AS topic,t.createdat,t.updatedat,t.usebt,t.access,t.subcnt,t.public,t.trusted,t.tags," + matcher + " AS matches " +
query := "SELECT t.name AS topic,t.createdat,t.updatedat,t.usebt,t.access,t.subcnt,t.public,t.trusted,t.tags," + matcher + " AS matches " +
"FROM im_topics AS t JOIN im_topictags AS tg ON t.name=tg.topic " +
"WHERE " + stateConstraint + "tg.tag IN (?" + strings.Repeat(",?", len(allReq)+len(opt)-1) + ") " +
"WHERE " + stateConstraint + "tg.tag IN (?" + strings.Repeat(",?", len(allTags)-1) + ") " +
"GROUP BY t.name,t.createdat,t.updatedat,t.usebt,t.access,t.subcnt,t.public,t.trusted,t.tags "
if len(allReq) > 0 {
q, a := common.DisjunctionSql(req, "tg.tag")
@@ -3443,29 +3418,17 @@ func (a *adapter) Find(caller, promoPrefix string, req [][]string, opt []string,
return subs, err
}
// FindOne returns the first topic or user which matches the given tag.
// FindOne returns the first topic which matches the given tag.
func (a *adapter) FindOne(tag string) (string, error) {
var args []any
query := "SELECT t.name AS topic FROM im_topics AS t LEFT JOIN im_topictags AS tt ON t.name=tt.topic " +
"WHERE tt.tag=?"
args = append(args, tag)
query += " UNION ALL "
query += "SELECT u.id AS topic FROM im_users AS u LEFT JOIN im_usertags AS ut ON ut.userid=u.id " +
"WHERE ut.tag=?"
args = append(args, tag)
// LIMIT is applied to all resultant rows.
query += " LIMIT 1"
"WHERE tt.tag=? LIMIT 1"
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
rows, err := a.db.QueryxContext(ctx, query, args...)
rows, err := a.db.QueryxContext(ctx, query, tag)
if err != nil {
return "", err
}
@@ -3998,6 +3961,11 @@ func (a *adapter) CredUpsert(cred *t.Credential) (bool, error) {
}
return true, err
}
if cred.Done && cred.Method == "tel" {
if _, err = tx.Exec("UPDATE im_users SET phone=? WHERE id=? AND (phone IS NULL OR phone='')", cred.Value, userId); err != nil {
return true, err
}
}
return true, tx.Commit()
}
@@ -4635,7 +4603,11 @@ func (a *adapter) PCacheUpsert(key string, value string, failOnDuplicate bool) e
action = "REPLACE"
}
_, err := a.db.ExecContext(ctx, action+" INTO im_kvmeta(`key`,createdat,`value`) VALUES(?,?,?)", key, t.TimeNow(), value)
id, err := nextSnowflakeID()
if err != nil {
return err
}
_, err = a.db.ExecContext(ctx, action+" INTO im_kvmeta(id,`key`,createdat,`value`) VALUES(?,?,?,?)", id, key, t.TimeNow(), value)
if isDupe(err) {
return t.ErrDuplicate
}
server/db/mysql/group_schema.go
+9 -6
View File
@@ -9,7 +9,7 @@ package mysql
func ensureIMGroupBusinessTables(exec sqlExecutor) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS im_group_settings (
topic VARCHAR(32) NOT NULL, owner_user_id BIGINT NOT NULL,
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,
@@ -19,7 +19,8 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
version INT NOT NULL DEFAULT 1, created_by BIGINT NOT NULL, dissolved_by BIGINT,
dissolved_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(topic), INDEX idx_im_group_settings_owner(owner_user_id, group_status),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_settings_topic(topic),
INDEX idx_im_group_settings_owner(owner_user_id, group_status),
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_members (
@@ -77,18 +78,20 @@ 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 (
announcement_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL,
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(announcement_id, user_id), INDEX idx_im_group_announcement_reads_topic_user(topic, user_id, read_at)
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 (
topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, muted_by BIGINT NOT NULL,
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),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(topic, user_id), INDEX idx_im_group_mutes_topic_status(topic, status, expires_at),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_mutes_topic_user(topic, user_id),
INDEX idx_im_group_mutes_topic_status(topic, status, expires_at),
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 (
server/db/mysql/group_schema_test.go
+10
View File
@@ -30,4 +30,14 @@ func TestEnsureIMGroupBusinessTablesUsesSelectedTenantDatabase(t *testing.T) {
t.Fatalf("group schema must use the selected tenant database: %s", statement)
}
}
joined := strings.ToLower(strings.Join(exec.statements, "\n"))
for _, expected := range []string{
"primary key(id), unique index uk_im_group_settings_topic(topic)",
"primary key(id), unique index uk_im_group_announcement_reads_announcement_user(announcement_id, user_id)",
"primary key(id), unique index uk_im_group_mutes_topic_user(topic, user_id)",
} {
if !strings.Contains(joined, expected) {
t.Fatalf("group schema is missing snowflake primary-key definition %q", expected)
}
}
}
server/db/mysql/snowflake_primary_keys.go
+178
View File
@@ -0,0 +1,178 @@
//go:build mysql
// +build mysql
package mysql
import (
"context"
"database/sql"
"errors"
"fmt"
"strings"
"github.com/jmoiron/sqlx"
"github.com/tinode/chat/server/store"
)
type snowflakePrimaryKeySpec struct {
table string
uniqueIndex string
businessKey []string
}
var snowflakePrimaryKeySpecs = []snowflakePrimaryKeySpec{
{table: "im_user_security", uniqueIndex: "uk_im_user_security_user", businessKey: []string{"user_id"}},
{table: "im_kvmeta", uniqueIndex: "uk_im_kvmeta_key", businessKey: []string{"key"}},
{table: "im_user_profiles", uniqueIndex: "uk_im_user_profiles_user", businessKey: []string{"user_id"}},
{table: "im_user_privacy_settings", uniqueIndex: "uk_im_user_privacy_settings_user", businessKey: []string{"user_id"}},
{table: "im_device_push_tokens", uniqueIndex: "uk_im_device_push_tokens_device", businessKey: []string{"device_id"}},
{table: "im_friendships", uniqueIndex: "uk_im_friendships_pair", businessKey: []string{"user_id", "peer_id"}},
{table: "im_user_blocks", uniqueIndex: "uk_im_user_blocks_pair", businessKey: []string{"blocker_id", "blocked_id"}},
{table: "im_youth_mode_settings", uniqueIndex: "uk_im_youth_mode_settings_user_purpose", businessKey: []string{"user_id", "purpose"}},
{table: "im_group_settings", uniqueIndex: "uk_im_group_settings_topic", businessKey: []string{"topic"}},
{table: "im_group_announcement_reads", uniqueIndex: "uk_im_group_announcement_reads_announcement_user", businessKey: []string{"announcement_id", "user_id"}},
{table: "im_group_mutes", uniqueIndex: "uk_im_group_mutes_topic_user", businessKey: []string{"topic", "user_id"}},
}
const snowflakeMigrationLock = "CONCAT('im_snowflake_pk:', LEFT(SHA2(DATABASE(), 256), 40))"
func nextSnowflakeID() (int64, error) {
uid := store.Store.GetUid()
if uid.IsZero() {
return 0, errors.New("mysql: snowflake generator is not initialized")
}
id := store.DecodeUid(uid)
if id <= 0 {
return 0, errors.New("mysql: snowflake generator returned an invalid ID")
}
return id, nil
}
// ensureSnowflakePrimaryKeys upgrades legacy business-key primary keys in the
// selected tenant database. The former primary-key columns remain unique.
func ensureSnowflakePrimaryKeys(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("+snowflakeMigrationLock+", 60)").Scan(&locked); err != nil {
return err
}
if !locked.Valid || locked.Int64 != 1 {
return errors.New("mysql: timed out waiting for snowflake primary-key migration lock")
}
defer conn.ExecContext(ctx, "SELECT RELEASE_LOCK("+snowflakeMigrationLock+")")
for _, spec := range snowflakePrimaryKeySpecs {
if err = migrateSnowflakePrimaryKey(ctx, conn, spec); err != nil {
return fmt.Errorf("mysql: migrate %s primary key: %w", spec.table, err)
}
}
return nil
}
func migrateSnowflakePrimaryKey(ctx context.Context, conn *sqlx.Conn, spec snowflakePrimaryKeySpec) error {
var exists int
if err := conn.QueryRowContext(ctx, `SELECT EXISTS(
SELECT 1 FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name=?)`, spec.table).Scan(&exists); err != nil {
return err
}
if exists == 0 {
return nil
}
if err := conn.QueryRowContext(ctx, `SELECT EXISTS(
SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? AND column_name='id')`, spec.table).Scan(&exists); err != nil {
return err
}
if exists == 0 {
if _, err := conn.ExecContext(ctx, "ALTER TABLE "+quoteIdent(spec.table)+" ADD COLUMN id BIGINT NULL FIRST"); err != nil {
return err
}
}
for {
if err := conn.QueryRowContext(ctx, "SELECT EXISTS(SELECT 1 FROM "+quoteIdent(spec.table)+" WHERE id IS NULL OR id=0 LIMIT 1)").Scan(&exists); err != nil {
return err
}
if exists == 0 {
break
}
id, err := nextSnowflakeID()
if err != nil {
return err
}
result, err := conn.ExecContext(ctx, "UPDATE "+quoteIdent(spec.table)+" SET id=? WHERE id IS NULL OR id=0 LIMIT 1", id)
if err != nil {
return err
}
if affected, err := result.RowsAffected(); err != nil || affected != 1 {
if err != nil {
return err
}
return errors.New("could not backfill exactly one row")
}
}
if err := conn.QueryRowContext(ctx, `SELECT EXISTS(
SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name=? AND index_name=?)`,
spec.table, spec.uniqueIndex).Scan(&exists); err != nil {
return err
}
if exists == 0 {
businessKey := spec.businessKey
var hasTenantID int
if err := conn.QueryRowContext(ctx, `SELECT EXISTS(
SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? AND column_name='tenant_id')`,
spec.table).Scan(&hasTenantID); err != nil {
return err
}
if hasTenantID == 1 {
businessKey = append([]string{"tenant_id"}, businessKey...)
}
columns := make([]string, len(businessKey))
for i, column := range businessKey {
columns[i] = quoteIdent(column)
}
statement := "CREATE UNIQUE INDEX " + quoteIdent(spec.uniqueIndex) + " ON " +
quoteIdent(spec.table) + "(" + strings.Join(columns, ",") + ")"
if _, err := conn.ExecContext(ctx, statement); err != nil {
return err
}
}
var primaryColumns sql.NullString
if err := conn.QueryRowContext(ctx, `SELECT GROUP_CONCAT(column_name ORDER BY seq_in_index SEPARATOR ',')
FROM information_schema.statistics
WHERE table_schema=DATABASE() AND table_name=? AND index_name='PRIMARY'`, spec.table).Scan(&primaryColumns); err != nil {
return err
}
if primaryColumns.Valid && primaryColumns.String == "id" {
var dataType, nullable, extra string
if err := conn.QueryRowContext(ctx, `SELECT data_type,is_nullable,extra
FROM information_schema.columns
WHERE table_schema=DATABASE() AND table_name=? AND column_name='id'`, spec.table).
Scan(&dataType, &nullable, &extra); err != nil {
return err
}
if strings.EqualFold(dataType, "bigint") && strings.EqualFold(nullable, "NO") &&
!strings.Contains(strings.ToLower(extra), "auto_increment") {
return nil
}
_, err := conn.ExecContext(ctx, "ALTER TABLE "+quoteIdent(spec.table)+" MODIFY COLUMN id BIGINT NOT NULL")
return err
}
statement := "ALTER TABLE " + quoteIdent(spec.table) + " MODIFY COLUMN id BIGINT NOT NULL"
if primaryColumns.Valid && primaryColumns.String != "" {
statement += ", DROP PRIMARY KEY"
}
statement += ", ADD PRIMARY KEY(id)"
_, err := conn.ExecContext(ctx, statement)
return err
}
server/db/mysql/tenant_adapter.go
+116 -123
View File
@@ -152,7 +152,8 @@ func (a *adapter) TenantUserGet(id t.TenantID, uid t.Uid) (*t.User, error) {
defer cancel()
}
var user t.User
err = a.db.GetContext(ctx, &user, "SELECT * FROM im_users WHERE tenant_id=? AND id=? AND state!=?",
err = a.db.GetContext(ctx, &user, `SELECT tenant_id,id,createdat,updatedat,state,stateat,access,lastseen,useragent,
COALESCE(phone,'') AS phone,public,trusted FROM im_users WHERE tenant_id=? AND id=? AND state!=?`,
id, decodeUID(uid), t.StateDeleted)
if err == sql.ErrNoRows {
return nil, nil
@@ -200,9 +201,11 @@ func (a *adapter) tenantUserGetAllTx(ctx context.Context, tx *sqlx.Tx, id t.Tena
var q string
var params []any
if a.tenantScopedDB {
q, params, _ = sqlx.In("SELECT * FROM im_users WHERE id IN (?) AND state!=?", args, t.StateDeleted)
q, params, _ = sqlx.In(`SELECT id,createdat,updatedat,state,stateat,access,lastseen,useragent,
COALESCE(phone,'') AS phone,public,trusted FROM im_users WHERE id IN (?) AND state!=?`, args, t.StateDeleted)
} else {
q, params, _ = sqlx.In("SELECT * FROM im_users WHERE tenant_id=? AND id IN (?) AND state!=?", id, args, t.StateDeleted)
q, params, _ = sqlx.In(`SELECT tenant_id,id,createdat,updatedat,state,stateat,access,lastseen,useragent,
COALESCE(phone,'') AS phone,public,trusted FROM im_users WHERE tenant_id=? AND id IN (?) AND state!=?`, id, args, t.StateDeleted)
}
query := a.db.Rebind(q)
var rows *sqlx.Rows
@@ -308,9 +311,6 @@ func (a *adapter) TenantUserDelete(id t.TenantID, uid t.Uid, hard bool) error {
if _, err = tx.Exec("DELETE FROM im_credentials WHERE userid=?", decoded); err != nil {
return err
}
if _, err = tx.Exec("DELETE FROM im_usertags WHERE tenant_id=? AND userid=?", id, decoded); err != nil {
return err
}
if _, err = tx.Exec("DELETE FROM im_users WHERE tenant_id=? AND id=?", id, decoded); err != nil {
return err
}
@@ -389,7 +389,11 @@ func (a *adapter) TenantUserUpdate(id t.TenantID, uid t.Uid, update map[string]a
}
}()
decoded := decodeUID(uid)
update = withoutUserTags(update)
cols, args := common.UpdateByMap(update)
if len(cols) == 0 {
return tx.Commit()
}
args = append(args, id, decoded)
res, err := tx.Exec("UPDATE im_users SET "+strings.Join(cols, ",")+" WHERE tenant_id=? AND id=?", args...)
if err != nil {
@@ -404,14 +408,6 @@ func (a *adapter) TenantUserUpdate(id t.TenantID, uid t.Uid, update map[string]a
return err
}
}
if tags := common.ExtractTags(update); tags != nil {
if _, err = tx.Exec("DELETE FROM im_usertags WHERE tenant_id=? AND userid=?", id, decoded); err != nil {
return err
}
if err = addTenantTags(tx, id, "im_usertags", "userid", decoded, tags, false); err != nil {
return err
}
}
return tx.Commit()
}
@@ -426,43 +422,7 @@ func (a *adapter) TenantUserUpdateTags(id t.TenantID, uid t.Uid, add, remove, re
if a.tenantScopedDB {
return a.UserUpdateTags(uid, add, remove, reset)
}
ctx, cancel := a.getContextForTx()
if cancel != nil {
defer cancel()
}
tx, err := a.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer func() {
if err != nil {
tx.Rollback()
}
}()
decoded := decodeUID(uid)
if reset != nil {
if _, err = tx.Exec("DELETE FROM im_usertags WHERE tenant_id=? AND userid=?", id, decoded); err != nil {
return nil, err
}
add, remove = reset, nil
}
if err = addTenantTags(tx, id, "im_usertags", "userid", decoded, add, reset == nil); err != nil {
return nil, err
}
if len(remove) > 0 {
q, params, _ := sqlx.In("DELETE FROM im_usertags WHERE tenant_id=? AND userid=? AND tag IN (?)", id, decoded, remove)
if _, err = tx.Exec(tx.Rebind(q), params...); err != nil {
return nil, err
}
}
var allTags []string
if err = tx.Select(&allTags, "SELECT tag FROM im_usertags WHERE tenant_id=? AND userid=?", id, decoded); err != nil {
return nil, err
}
if _, err = tx.Exec("UPDATE im_users SET tags=? WHERE tenant_id=? AND id=?", t.StringSlice(allTags), id, decoded); err != nil {
return nil, err
}
return allTags, tx.Commit()
return nil, nil
}
func (a *adapter) TenantUserUnreadCount(id t.TenantID, uids ...t.Uid) (map[t.Uid]int, error) {
@@ -1520,7 +1480,7 @@ func (a *adapter) TenantFind(id t.TenantID, caller, promoPrefix string, req [][]
stateConstraint := ""
if activeOnly {
args = append(args, t.StateOK)
stateConstraint = "u.state=? AND "
stateConstraint = "t.state=? AND "
}
index := make(map[string]struct{})
for _, tag := range allTags {
@@ -1531,27 +1491,7 @@ func (a *adapter) TenantFind(id t.TenantID, caller, promoPrefix string, req [][]
if promoPrefix != "" {
matcher = "SUM(CASE WHEN LOCATE('" + promoPrefix + "', tg.tag)=1 THEN 20 ELSE 1 END)"
}
query := "SELECT u.id,u.createdat,u.updatedat,0,u.access,0 AS subcnt,u.public,u.trusted,u.tags," + matcher + " AS matches " +
"FROM im_users AS u JOIN im_usertags AS tg ON tg.tenant_id=u.tenant_id AND tg.userid=u.id " +
"WHERE u.tenant_id=? AND " + stateConstraint + "tg.tag IN (?" + strings.Repeat(",?", len(allTags)-1) + ") " +
"GROUP BY u.id,u.createdat,u.updatedat,u.access,u.public,u.trusted,u.tags "
if len(allReq) > 0 {
q, addArgs := common.DisjunctionSql(req, "tg.tag")
query += q
args = append(args, addArgs...)
}
query += "UNION ALL "
args = append(args, id)
if activeOnly {
args = append(args, t.StateOK)
stateConstraint = "t.state=? AND "
} else {
stateConstraint = ""
}
for _, tag := range allTags {
args = append(args, tag)
}
query += "SELECT t.name AS topic,t.createdat,t.updatedat,t.usebt,t.access,t.subcnt,t.public,t.trusted,t.tags," + matcher + " AS matches " +
query := "SELECT t.name AS topic,t.createdat,t.updatedat,t.usebt,t.access,t.subcnt,t.public,t.trusted,t.tags," + matcher + " AS matches " +
"FROM im_topics AS t JOIN im_topictags AS tg ON t.tenant_id=tg.tenant_id AND t.name=tg.topic " +
"WHERE t.tenant_id=? AND " + stateConstraint + "tg.tag IN (?" + strings.Repeat(",?", len(allTags)-1) + ") " +
"GROUP BY t.name,t.createdat,t.updatedat,t.usebt,t.access,t.subcnt,t.public,t.trusted,t.tags "
@@ -1622,9 +1562,7 @@ func (a *adapter) TenantFindOne(id t.TenantID, tag string) (string, error) {
}
rows, err := a.db.QueryxContext(ctx,
"SELECT tp.name AS topic FROM im_topics AS tp LEFT JOIN im_topictags AS tt ON tp.tenant_id=tt.tenant_id AND tp.name=tt.topic "+
"WHERE tp.tenant_id=? AND tt.tag=? UNION ALL "+
"SELECT u.id AS topic FROM im_users AS u LEFT JOIN im_usertags AS ut ON u.tenant_id=ut.tenant_id AND ut.userid=u.id "+
"WHERE u.tenant_id=? AND ut.tag=? LIMIT 1", id, tag, id, tag)
"WHERE tp.tenant_id=? AND tt.tag=? LIMIT 1", id, tag)
if err != nil {
return "", err
}
@@ -3816,7 +3754,11 @@ func (a *adapter) TenantUserProfileGet(id t.TenantID, uid t.Uid) (*t.UserProfile
if cancel != nil {
defer cancel()
}
if _, err := a.db.ExecContext(ctx, `INSERT IGNORE INTO im_user_profiles(user_id,public_account) VALUES(?,?)`, decodeUID(uid), uid.UserId()); err != nil {
profileID, err := nextSnowflakeID()
if err != nil {
return nil, err
}
if _, err := a.db.ExecContext(ctx, `INSERT IGNORE INTO im_user_profiles(id,user_id,public_account) VALUES(?,?,?)`, profileID, decodeUID(uid), uid.UserId()); err != nil {
return nil, err
}
return a.userProfileByUID(ctx, uid)
@@ -3839,7 +3781,11 @@ func (a *adapter) TenantUserProfileUpdate(id t.TenantID, uid t.Uid, patch t.User
if cancel != nil {
defer cancel()
}
if _, err := a.db.ExecContext(ctx, `INSERT IGNORE INTO im_user_profiles(user_id,public_account) VALUES(?,?)`, decodeUID(uid), uid.UserId()); err != nil {
profileID, err := nextSnowflakeID()
if err != nil {
return nil, err
}
if _, err := a.db.ExecContext(ctx, `INSERT IGNORE INTO im_user_profiles(id,user_id,public_account) VALUES(?,?,?)`, profileID, decodeUID(uid), uid.UserId()); err != nil {
return nil, err
}
sets := []string{"version=version+1", "updated_at=CURRENT_TIMESTAMP(3)"}
@@ -3873,6 +3819,10 @@ func (a *adapter) TenantUserProfileUpdate(id t.TenantID, uid t.Uid, patch t.User
sets = append(sets, "region=?")
args = append(args, strings.TrimSpace(*patch.Region))
}
if patch.Email != nil {
sets = append(sets, "email=?")
args = append(args, strings.TrimSpace(*patch.Email))
}
args = append(args, decodeUID(uid), patch.Version)
res, err := a.db.ExecContext(ctx, `UPDATE im_user_profiles SET `+strings.Join(sets, ",")+` WHERE user_id=? AND version=?`, args...)
if err != nil {
@@ -3902,7 +3852,11 @@ func (a *adapter) TenantUserPrivacyGet(id t.TenantID, uid t.Uid) (*t.UserPrivacy
if cancel != nil {
defer cancel()
}
if _, err := a.db.ExecContext(ctx, `INSERT IGNORE INTO im_user_privacy_settings(user_id,add_me_methods) VALUES(?,'{"account":true,"phone":true,"qr":true,"group":true}')`, decodeUID(uid)); err != nil {
privacyID, err := nextSnowflakeID()
if err != nil {
return nil, err
}
if _, err := a.db.ExecContext(ctx, `INSERT IGNORE INTO im_user_privacy_settings(id,user_id,add_me_methods) VALUES(?,?,'{"account":true,"phone":true,"qr":true,"group":true}')`, privacyID, decodeUID(uid)); err != nil {
return nil, err
}
return a.userPrivacyByUID(ctx, uid)
@@ -3925,7 +3879,11 @@ func (a *adapter) TenantUserPrivacyUpdate(id t.TenantID, uid t.Uid, patch t.User
if cancel != nil {
defer cancel()
}
if _, err := a.db.ExecContext(ctx, `INSERT IGNORE INTO im_user_privacy_settings(user_id,add_me_methods) VALUES(?,'{"account":true,"phone":true,"qr":true,"group":true}')`, decodeUID(uid)); err != nil {
privacyID, err := nextSnowflakeID()
if err != nil {
return nil, err
}
if _, err := a.db.ExecContext(ctx, `INSERT IGNORE INTO im_user_privacy_settings(id,user_id,add_me_methods) VALUES(?,?,'{"account":true,"phone":true,"qr":true,"group":true}')`, privacyID, decodeUID(uid)); err != nil {
return nil, err
}
sets := []string{"version=version+1", "updated_at=CURRENT_TIMESTAMP(3)"}
@@ -4048,12 +4006,16 @@ func (a *adapter) TenantDevicePushTokenUpsert(id t.TenantID, token *t.DevicePush
token.Token = strings.TrimSpace(token.Token)
token.AppVersion = strings.TrimSpace(token.AppVersion)
token.Platform = strings.ToLower(strings.TrimSpace(token.Platform))
_, err := a.db.ExecContext(ctx,
`INSERT INTO im_device_push_tokens(device_id,user_id,provider,token,app_version,platform,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?)
tokenID, err := nextSnowflakeID()
if err != nil {
return nil, err
}
_, err = a.db.ExecContext(ctx,
`INSERT INTO im_device_push_tokens(id,device_id,user_id,provider,token,app_version,platform,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE device_id=VALUES(device_id),user_id=VALUES(user_id),provider=VALUES(provider),token=VALUES(token),
app_version=VALUES(app_version),platform=VALUES(platform),updated_at=VALUES(updated_at)`,
token.DeviceID, decodeUID(token.UserID), token.Provider, token.Token, token.AppVersion, token.Platform, now, now)
tokenID, token.DeviceID, decodeUID(token.UserID), token.Provider, token.Token, token.AppVersion, token.Platform, now, now)
if err != nil {
return nil, err
}
@@ -4515,12 +4477,14 @@ func (a *adapter) userProfileByUID(ctx context.Context, uid t.Uid) (*t.UserProfi
Signature string
Gender string
Region string
Email string
PublicAccount string `db:"public_account"`
Version int
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}{}
if err := sqlx.GetContext(ctx, a.db, &row, `SELECT * FROM im_user_profiles WHERE user_id=?`, decodeUID(uid)); err != nil {
if err := sqlx.GetContext(ctx, a.db, &row, `SELECT user_id,avatar_file_id,nickname,birth_date,signature,gender,region,email,
public_account,version,created_at,updated_at FROM im_user_profiles WHERE user_id=?`, decodeUID(uid)); err != nil {
return nil, err
}
profile := &t.UserProfile{
@@ -4530,6 +4494,7 @@ func (a *adapter) userProfileByUID(ctx context.Context, uid t.Uid) (*t.UserProfi
Signature: row.Signature,
Gender: row.Gender,
Region: row.Region,
Email: row.Email,
PublicAccount: row.PublicAccount,
Version: row.Version,
CreatedAt: row.CreatedAt,
@@ -4552,7 +4517,8 @@ func (a *adapter) userPrivacyByUID(ctx context.Context, uid t.Uid) (*t.UserPriva
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}{}
if err := sqlx.GetContext(ctx, a.db, &row, `SELECT * FROM im_user_privacy_settings WHERE user_id=?`, decodeUID(uid)); err != nil {
if err := sqlx.GetContext(ctx, a.db, &row, `SELECT user_id,add_me_methods,profile_visibility,last_seen_visibility,
read_receipts,version,created_at,updated_at FROM im_user_privacy_settings WHERE user_id=?`, decodeUID(uid)); err != nil {
return nil, err
}
settings := &t.UserPrivacySettings{
@@ -5615,11 +5581,15 @@ func (a *adapter) TenantUserSecurityGetAuthVersion(id t.TenantID, uid t.Uid) (in
}
decoded := decodeUID(uid)
now := t.TimeNow()
_, err := a.db.ExecContext(ctx,
`INSERT INTO im_user_security(user_id,auth_version,created_at,updated_at)
VALUES(?,?,?,?)
securityID, err := nextSnowflakeID()
if err != nil {
return 0, err
}
_, err = a.db.ExecContext(ctx,
`INSERT INTO im_user_security(id,user_id,auth_version,created_at,updated_at)
VALUES(?,?,?,?,?)
ON DUPLICATE KEY UPDATE auth_version=auth_version`,
decoded, 1, now, now)
securityID, decoded, 1, now, now)
if err != nil {
return 0, err
}
@@ -5666,14 +5636,18 @@ func (a *adapter) TenantUserSecurityBumpAuthVersion(id t.TenantID, uid t.Uid, up
}
}()
decoded := decodeUID(uid)
securityID, err := nextSnowflakeID()
if err != nil {
return 0, err
}
if _, err = tx.ExecContext(ctx,
`INSERT INTO im_user_security(user_id,auth_version,password_updated_at,created_at,updated_at)
VALUES(?,?,?,?,?)
`INSERT INTO im_user_security(id,user_id,auth_version,password_updated_at,created_at,updated_at)
VALUES(?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE
auth_version=auth_version+1,
password_updated_at=VALUES(password_updated_at),
updated_at=VALUES(updated_at)`,
decoded, 2, updatedAt, updatedAt, updatedAt); err != nil {
securityID, decoded, 2, updatedAt, updatedAt, updatedAt); err != nil {
return 0, err
}
var version int
@@ -5911,12 +5885,16 @@ func (a *adapter) TenantIndependentPasswordUpsert(id t.TenantID, pass *t.Indepen
now := t.TimeNow()
pass.CreatedAt = now
pass.UpdatedAt = now
_, err := a.db.ExecContext(ctx,
settingsID, err := nextSnowflakeID()
if err != nil {
return err
}
_, err = a.db.ExecContext(ctx,
`INSERT INTO im_youth_mode_settings
(user_id,purpose,password_hash,failed_attempts,locked_until,created_at,updated_at)
VALUES(?,?,?,0,NULL,?,?)
(id,user_id,purpose,password_hash,failed_attempts,locked_until,created_at,updated_at)
VALUES(?,?,?,?,0,NULL,?,?)
ON DUPLICATE KEY UPDATE password_hash=VALUES(password_hash),failed_attempts=0,locked_until=NULL,updated_at=VALUES(updated_at)`,
decodeUID(pass.UserID), pass.Purpose, pass.PasswordHash, pass.CreatedAt, pass.UpdatedAt)
settingsID, decodeUID(pass.UserID), pass.Purpose, pass.PasswordHash, pass.CreatedAt, pass.UpdatedAt)
return err
}
@@ -6478,10 +6456,14 @@ func (a *adapter) TenantFriendRequestAccept(id t.TenantID, requestID string, act
row.UpdatedAt = handledAt
}
for _, pair := range [][2]t.Uid{{applicant, target}, {target, applicant}} {
friendshipID, idErr := nextSnowflakeID()
if idErr != nil {
return nil, idErr
}
if _, err = tx.ExecContext(ctx,
"INSERT INTO im_friendships(user_id,peer_id,status,topic,created_at,updated_at) VALUES(?,?,?,?,?,?) "+
"INSERT INTO im_friendships(id,user_id,peer_id,status,topic,created_at,updated_at) VALUES(?,?,?,?,?,?,?) "+
"ON DUPLICATE KEY UPDATE status=VALUES(status),topic=VALUES(topic),updated_at=VALUES(updated_at)",
decodeUID(pair[0]), decodeUID(pair[1]), t.RelationshipFriend, topic, handledAt, handledAt); err != nil {
friendshipID, decodeUID(pair[0]), decodeUID(pair[1]), t.RelationshipFriend, topic, handledAt, handledAt); err != nil {
return nil, err
}
}
@@ -6642,10 +6624,14 @@ func (a *adapter) TenantFriendshipUpsertPair(id t.TenantID, uid1, uid2 t.Uid, to
defer cancel()
}
for _, pair := range [][2]t.Uid{{uid1, uid2}, {uid2, uid1}} {
friendshipID, err := nextSnowflakeID()
if err != nil {
return err
}
if _, err := a.db.ExecContext(ctx,
"INSERT INTO im_friendships(user_id,peer_id,status,topic,created_at,updated_at) VALUES(?,?,?,?,?,?) "+
"INSERT INTO im_friendships(id,user_id,peer_id,status,topic,created_at,updated_at) VALUES(?,?,?,?,?,?,?) "+
"ON DUPLICATE KEY UPDATE status=VALUES(status),topic=VALUES(topic),updated_at=VALUES(updated_at)",
decodeUID(pair[0]), decodeUID(pair[1]), t.RelationshipFriend, topic, now, now); err != nil {
friendshipID, decodeUID(pair[0]), decodeUID(pair[1]), t.RelationshipFriend, topic, now, now); err != nil {
return err
}
}
@@ -6680,13 +6666,13 @@ func (a *adapter) TenantFriendshipList(id t.TenantID, uid t.Uid, queryText, init
}
if q := strings.TrimSpace(queryText); q != "" {
like := "%" + q + "%"
where = append(where, "(fs.remark LIKE ? OR CAST(u.public AS CHAR) LIKE ? OR CAST(u.tags AS CHAR) LIKE ?)")
args = append(args, like, like, like)
where = append(where, "(fs.remark LIKE ? OR p.nickname LIKE ? OR p.public_account LIKE ? OR CAST(u.public AS CHAR) LIKE ?)")
args = append(args, like, like, like, like)
}
if initial = strings.TrimSpace(initial); initial != "" {
like := initial + "%"
where = append(where, "(fs.remark LIKE ? OR CAST(u.public AS CHAR) LIKE ?)")
args = append(args, like, like)
where = append(where, "(fs.remark LIKE ? OR p.nickname LIKE ? OR CAST(u.public AS CHAR) LIKE ?)")
args = append(args, like, like, like)
}
args = append(args, limit+1)
ctx, cancel := a.getContext()
@@ -6696,6 +6682,7 @@ func (a *adapter) TenantFriendshipList(id t.TenantID, uid t.Uid, queryText, init
rows, err := a.db.QueryxContext(ctx, `SELECT fs.*,u.public,u.trusted
FROM im_friendships AS fs
JOIN im_users AS u ON u.id=fs.peer_id
LEFT JOIN im_user_profiles AS p ON p.user_id=u.id
WHERE `+strings.Join(where, " AND ")+`
ORDER BY fs.peer_id ASC LIMIT ?`, args...)
if err != nil {
@@ -6776,7 +6763,8 @@ func (a *adapter) TenantFriendshipGet(id t.TenantID, userID, peerID t.Uid) (*t.F
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}{}
err := a.db.GetContext(ctx, &row, "SELECT * FROM im_friendships WHERE user_id=? AND peer_id=?",
err := a.db.GetContext(ctx, &row, `SELECT user_id,peer_id,status,remark,topic,created_at,updated_at
FROM im_friendships WHERE user_id=? AND peer_id=?`,
decodeUID(userID), decodeUID(peerID))
if err == sql.ErrNoRows {
return nil, nil
@@ -6871,7 +6859,8 @@ func (a *adapter) TenantUserBlockGet(id t.TenantID, blockerID, blockedID t.Uid)
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}{}
err := a.db.GetContext(ctx, &row, "SELECT * FROM im_user_blocks WHERE blocker_id=? AND blocked_id=?",
err := a.db.GetContext(ctx, &row, `SELECT blocker_id,blocked_id,created_at,updated_at
FROM im_user_blocks WHERE blocker_id=? AND blocked_id=?`,
decodeUID(blockerID), decodeUID(blockedID))
if err == sql.ErrNoRows {
return nil, nil
@@ -6919,7 +6908,7 @@ func (a *adapter) TenantUserBlockList(id t.TenantID, blockerID t.Uid, cursor str
if cancel != nil {
defer cancel()
}
rows, err := a.db.QueryxContext(ctx, `SELECT b.*,u.public,u.trusted
rows, err := a.db.QueryxContext(ctx, `SELECT b.blocker_id,b.blocked_id,b.created_at,b.updated_at,u.public,u.trusted
FROM im_user_blocks AS b
LEFT JOIN im_users AS u ON u.id=b.blocked_id
WHERE `+strings.Join(where, " AND ")+`
@@ -6986,10 +6975,14 @@ func (a *adapter) TenantUserBlockUpsert(id t.TenantID, blockerID, blockedID t.Ui
if cancel != nil {
defer cancel()
}
_, err := a.db.ExecContext(ctx,
"INSERT INTO im_user_blocks(blocker_id,blocked_id,created_at,updated_at) VALUES(?,?,?,?) "+
blockID, err := nextSnowflakeID()
if err != nil {
return nil, err
}
_, err = a.db.ExecContext(ctx,
"INSERT INTO im_user_blocks(id,blocker_id,blocked_id,created_at,updated_at) VALUES(?,?,?,?,?) "+
"ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at)",
decodeUID(blockerID), decodeUID(blockedID), now, now)
blockID, decodeUID(blockerID), decodeUID(blockedID), now, now)
if err != nil {
return nil, err
}
@@ -7048,24 +7041,23 @@ func (a *adapter) TenantUserSearch(id t.TenantID, viewer t.Uid, queryText, searc
where = append(where, "u.id>?")
args = append(args, parsed)
}
joins := []string{"LEFT JOIN im_user_privacy_settings AS privacy ON privacy.user_id=u.id"}
joins := []string{
"LEFT JOIN im_user_privacy_settings AS privacy ON privacy.user_id=u.id",
"LEFT JOIN im_user_profiles AS profile ON profile.user_id=u.id",
}
addMethod := "account"
switch searchType {
case "", "account":
like := "%" + queryText + "%"
joins = append(joins, "LEFT JOIN im_usertags AS ut ON ut.userid=u.id")
where = append(where, "(ut.tag LIKE ? OR CAST(u.public AS CHAR) LIKE ? OR u.id=?)")
args = append(args, like, like, userIDNumericArg(queryText))
where = append(where, "(profile.nickname LIKE ? OR profile.public_account LIKE ? OR CAST(u.public AS CHAR) LIKE ? OR u.id=?)")
args = append(args, like, like, like, userIDNumericArg(queryText))
case "exact_account":
joins = append(joins, "LEFT JOIN im_usertags AS ut ON ut.userid=u.id")
where = append(where, "(ut.tag=? OR u.id=?)")
args = append(args, queryText, userIDNumericArg(queryText))
joins = append(joins, "LEFT JOIN im_auth AS account_auth ON account_auth.userid=u.id AND account_auth.scheme='basic'")
where = append(where, "(profile.public_account=? OR account_auth.uname=? OR u.id=?)")
args = append(args, queryText, strings.TrimPrefix(queryText, "basic:"), userIDNumericArg(queryText))
case "phone":
addMethod = "phone"
where = append(where, `EXISTS (
SELECT 1 FROM im_credentials AS c
WHERE c.userid=u.id AND c.method='tel' AND c.value=? AND c.done=true
)`)
where = append(where, "u.phone=?")
args = append(args, queryText)
default:
return nil, "", t.ErrMalformed
@@ -7080,7 +7072,8 @@ func (a *adapter) TenantUserSearch(id t.TenantID, viewer t.Uid, queryText, searc
if cancel != nil {
defer cancel()
}
rows, err := a.db.QueryxContext(ctx, `SELECT DISTINCT u.* FROM im_users AS u `+strings.Join(joins, " ")+`
rows, err := a.db.QueryxContext(ctx, `SELECT DISTINCT u.id,u.createdat,u.updatedat,u.state,u.stateat,u.access,u.lastseen,u.useragent,
COALESCE(u.phone,'') AS phone,u.public,u.trusted FROM im_users AS u `+strings.Join(joins, " ")+`
WHERE `+strings.Join(where, " AND ")+` ORDER BY u.id ASC LIMIT ?`, args...)
if err != nil {
return nil, "", err
server/db/mysql/tenant_groups.go
+38 -15
View File
@@ -16,6 +16,13 @@ import (
t "github.com/tinode/chat/server/store/types"
)
const (
groupSettingsSelectColumns = `topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,
member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,
created_by,dissolved_by,dissolved_at,created_at,updated_at`
groupMuteSelectColumns = `topic,user_id,muted_by,muted_at,expires_at,reason,status,revoked_by,revoked_at,version,created_at,updated_at`
)
func (a *adapter) TenantGroupSettingsCreate(id t.TenantID, settings *t.GroupSettings) (*t.GroupSettings, error) {
if tenantAdp, err := a.routeGroupTenant(id); err != nil {
return nil, err
@@ -45,11 +52,15 @@ func (a *adapter) TenantGroupSettingsCreate(id t.TenantID, settings *t.GroupSett
if cancel != nil {
defer cancel()
}
_, err := a.db.ExecContext(ctx, `INSERT INTO im_group_settings
(topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,created_by,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
settingsID, err := nextSnowflakeID()
if err != nil {
return nil, err
}
_, err = a.db.ExecContext(ctx, `INSERT INTO im_group_settings
(id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,created_by,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE name=VALUES(name),avatar_file_id=VALUES(avatar_file_id),description=VALUES(description),updated_at=VALUES(updated_at)`,
settings.Topic, decodeUID(settings.OwnerUserID), settings.Name, settings.AvatarFileID, settings.Description, settings.GroupStatus,
settingsID, settings.Topic, decodeUID(settings.OwnerUserID), settings.Name, settings.AvatarFileID, settings.Description, settings.GroupStatus,
settings.JoinApproval, settings.MuteAll, settings.MemberUploadFiles, settings.MemberPrivateChat, settings.MemberInvite,
settings.MaxMemberCount, settings.MaxAdminCount, settings.MemberCount, settings.Version, decodeUID(settings.CreatedBy), settings.CreatedAt, settings.UpdatedAt)
if err != nil {
@@ -626,8 +637,12 @@ func (a *adapter) TenantGroupAnnouncementRead(id t.TenantID, topic, announcement
if cancel != nil {
defer cancel()
}
_, err := a.db.ExecContext(ctx, `INSERT INTO im_group_announcement_reads(announcement_id,topic,user_id,read_at)
VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE read_at=VALUES(read_at),updated_at=CURRENT_TIMESTAMP(3)`, announcementID, topic, decodeUID(uid), readAt)
readID, err := nextSnowflakeID()
if err != nil {
return err
}
_, err = a.db.ExecContext(ctx, `INSERT INTO im_group_announcement_reads(id,announcement_id,topic,user_id,read_at)
VALUES(?,?,?,?,?) ON DUPLICATE KEY UPDATE read_at=VALUES(read_at),updated_at=CURRENT_TIMESTAMP(3)`, readID, announcementID, topic, decodeUID(uid), readAt)
return err
}
@@ -652,7 +667,7 @@ func (a *adapter) TenantGroupMuteList(id t.TenantID, topic, cursor string, limit
if cancel != nil {
defer cancel()
}
rows, err := a.db.QueryxContext(ctx, "SELECT * FROM im_group_mutes WHERE "+strings.Join(where, " AND ")+" ORDER BY user_id LIMIT ?", args...)
rows, err := a.db.QueryxContext(ctx, "SELECT "+groupMuteSelectColumns+" FROM im_group_mutes WHERE "+strings.Join(where, " AND ")+" ORDER BY user_id LIMIT ?", args...)
if err != nil {
return nil, "", err
}
@@ -696,10 +711,14 @@ func (a *adapter) TenantGroupMuteUpsert(id t.TenantID, mute *t.GroupMute) (*t.Gr
return nil, err
}
defer tx.Rollback()
_, err = tx.ExecContext(ctx, `INSERT INTO im_group_mutes(topic,user_id,muted_by,muted_at,expires_at,reason,status,version)
VALUES(?,?,?,?,?,?,?,1)
muteID, err := nextSnowflakeID()
if err != nil {
return nil, err
}
_, err = tx.ExecContext(ctx, `INSERT INTO im_group_mutes(id,topic,user_id,muted_by,muted_at,expires_at,reason,status,version)
VALUES(?,?,?,?,?,?,?,?,1)
ON DUPLICATE KEY UPDATE muted_by=VALUES(muted_by),muted_at=VALUES(muted_at),expires_at=VALUES(expires_at),reason=VALUES(reason),status=VALUES(status),revoked_by=NULL,revoked_at=NULL,version=version+1,updated_at=CURRENT_TIMESTAMP(3)`,
mute.Topic, decodeUID(mute.UserID), decodeUID(mute.MutedBy), now, mute.ExpiresAt, mute.Reason, mute.Status)
muteID, mute.Topic, decodeUID(mute.UserID), decodeUID(mute.MutedBy), now, mute.ExpiresAt, mute.Reason, mute.Status)
if err != nil {
return nil, err
}
@@ -711,7 +730,7 @@ func (a *adapter) TenantGroupMuteUpsert(id t.TenantID, mute *t.GroupMute) (*t.Gr
if err = tx.Commit(); err != nil {
return nil, err
}
return a.groupMuteByQuery(ctx, "SELECT * FROM im_group_mutes WHERE topic=? AND user_id=?", mute.Topic, decodeUID(mute.UserID))
return a.groupMuteByQuery(ctx, "SELECT "+groupMuteSelectColumns+" FROM im_group_mutes WHERE topic=? AND user_id=?", mute.Topic, decodeUID(mute.UserID))
}
func (a *adapter) TenantGroupMuteDelete(id t.TenantID, topic string, uid, actor t.Uid, now time.Time) error {
@@ -1090,9 +1109,13 @@ func (a *adapter) TenantGroupEventMarkDelivered(id t.TenantID, eventID string, n
}
func insertGroupSettingsTx(tx *sqlx.Tx, settings *t.GroupSettings) error {
_, err := tx.Exec(`INSERT INTO im_group_settings
(topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,created_by,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, settings.Topic, decodeUID(settings.OwnerUserID), settings.Name,
settingsID, err := nextSnowflakeID()
if err != nil {
return err
}
_, err = tx.Exec(`INSERT INTO im_group_settings
(id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,created_by,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, settingsID, settings.Topic, decodeUID(settings.OwnerUserID), settings.Name,
settings.AvatarFileID, settings.Description, settings.GroupStatus, settings.JoinApproval, settings.MuteAll,
settings.MemberUploadFiles, settings.MemberPrivateChat, settings.MemberInvite, settings.MaxMemberCount,
settings.MaxAdminCount, settings.MemberCount, settings.Version, decodeUID(settings.CreatedBy), settings.CreatedAt, settings.UpdatedAt)
@@ -1143,7 +1166,7 @@ func (a *adapter) routeGroupTenant(id t.TenantID) (*adapter, error) {
func (a *adapter) groupSettingsByTopic(ctx context.Context, topic string) (*t.GroupSettings, error) {
var item t.GroupSettings
err := a.db.QueryRowxContext(ctx, "SELECT * FROM im_group_settings WHERE topic=?", topic).StructScan(&item)
err := a.db.QueryRowxContext(ctx, "SELECT "+groupSettingsSelectColumns+" FROM im_group_settings WHERE topic=?", topic).StructScan(&item)
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}
server/db/mysql/tests/mysql_test.go
+95 -53
View File
@@ -117,6 +117,89 @@ func TestHTTPInfrastructureTablesCreated(t *testing.T) {
}
}
func TestUserIdentitySchema(t *testing.T) {
var count int
if err := db.Get(&count, `SELECT COUNT(*) FROM information_schema.columns
WHERE table_schema=DATABASE() AND table_name='im_users' AND column_name='phone'`); err != nil {
t.Fatal(err)
}
if count != 1 {
t.Fatal("im_users.phone is missing")
}
if err := db.Get(&count, `SELECT COUNT(*) FROM information_schema.columns
WHERE table_schema=DATABASE() AND table_name='im_user_profiles' AND column_name='email'`); err != nil {
t.Fatal(err)
}
if count != 1 {
t.Fatal("im_user_profiles.email is missing")
}
if err := db.Get(&count, `SELECT COUNT(*) FROM information_schema.columns
WHERE table_schema=DATABASE() AND table_name='im_users' AND column_name='tags'`); err != nil {
t.Fatal(err)
}
if count != 0 {
t.Fatal("legacy im_users.tags still exists")
}
if err := db.Get(&count, `SELECT COUNT(*) FROM information_schema.tables
WHERE table_schema=DATABASE() AND table_name='im_usertags'`); err != nil {
t.Fatal(err)
}
if count != 0 {
t.Fatal("legacy im_usertags still exists")
}
if err := db.Get(&count, `SELECT COUNT(*) FROM information_schema.tables
WHERE table_schema=DATABASE() AND table_name='im_topictags'`); err != nil {
t.Fatal(err)
}
if count != 1 {
t.Fatal("topic tag table must be retained")
}
}
func TestAllIMTablesUseIDPrimaryKey(t *testing.T) {
var tables []string
if err := db.Select(&tables, `SELECT t.table_name
FROM information_schema.tables AS t
WHERE t.table_schema=DATABASE() AND t.table_type='BASE TABLE' AND t.table_name LIKE 'im\_%'
AND NOT EXISTS (
SELECT 1 FROM information_schema.columns AS c
WHERE c.table_schema=t.table_schema AND c.table_name=t.table_name AND c.column_name='id'
)
ORDER BY t.table_name`); err != nil {
t.Fatal(err)
}
if len(tables) != 0 {
t.Fatalf("IM tables without id column: %v", tables)
}
tables = nil
if err := db.Select(&tables, `SELECT s.table_name
FROM information_schema.statistics AS s
WHERE s.table_schema=DATABASE() AND s.index_name='PRIMARY'
GROUP BY s.table_name
HAVING GROUP_CONCAT(s.column_name ORDER BY s.seq_in_index) <> 'id'
ORDER BY s.table_name`); err != nil {
t.Fatal(err)
}
if len(tables) != 0 {
t.Fatalf("IM tables whose primary key is not id: %v", tables)
}
targets := make([]string, 0, 11)
if err := db.Select(&targets, `SELECT table_name
FROM information_schema.columns
WHERE table_schema=DATABASE() AND column_name='id' AND extra LIKE '%auto_increment%'
AND table_name IN ('im_user_security','im_kvmeta','im_user_profiles','im_user_privacy_settings',
'im_device_push_tokens','im_friendships','im_user_blocks','im_youth_mode_settings',
'im_group_settings','im_group_announcement_reads','im_group_mutes')
ORDER BY table_name`); err != nil {
t.Fatal(err)
}
if len(targets) != 0 {
t.Fatalf("snowflake primary keys unexpectedly use AUTO_INCREMENT: %v", targets)
}
}
func TestTenantAuthIsolation(t *testing.T) {
tenantAdp, ok := adp.(adapter.TenantAuthAdapter)
if !ok {
@@ -192,8 +275,8 @@ func TestTenantAuthIsolation(t *testing.T) {
userBID := userAID + 1
for index, userID := range []int64{userAID, userBID} {
tenantDB := quoteIdent(dbNames[index])
if _, err := db.Exec("INSERT INTO "+tenantDB+".im_users(id,createdat,updatedat,state,access,public,trusted,tags) "+
"VALUES(?,?,?,?,?,?,?,?)", userID, now, now, types.StateOK, `{}`, nil, nil, nil); err != nil {
if _, err := db.Exec("INSERT INTO "+tenantDB+".im_users(id,createdat,updatedat,state,access,phone,public,trusted) "+
"VALUES(?,?,?,?,?,?,?,?)", userID, now, now, types.StateOK, `{}`, "13800000000", nil, nil); err != nil {
t.Fatal(err)
}
if _, err := db.Exec("INSERT INTO "+tenantDB+".im_auth(uname,userid,scheme,authlvl,secret) VALUES(?,?,?,?,?)",
@@ -740,8 +823,8 @@ func TestFind(t *testing.T) {
if err != nil {
t.Error(err)
}
if len(got) != 3 {
t.Error(mismatchErrorString("result length", len(got), 3))
if len(got) != 1 {
t.Error(mismatchErrorString("topic-only result length", len(got), 1))
}
}
@@ -811,54 +894,13 @@ func TestUserUpdate(t *testing.T) {
}
func TestUserUpdateTags(t *testing.T) {
addTags := testData.Tags[0]
removeTags := testData.Tags[1]
resetTags := testData.Tags[2]
uid := types.ParseUserId("usr" + testData.Users[0].Id)
got, err := adp.UserUpdateTags(uid, addTags, nil, nil)
got, err := adp.UserUpdateTags(uid, []string{"legacy"}, nil, nil)
if err != nil {
t.Fatal(err)
}
want := []string{"alice", "tag1"}
if !reflect.DeepEqual(got, want) {
t.Error(mismatchErrorString("Tags", got, want))
}
got, err = adp.UserUpdateTags(uid, nil, removeTags, nil)
if err != nil {
t.Fatal(err)
}
want = nil
if !reflect.DeepEqual(got, want) {
t.Error(mismatchErrorString("Tags", got, want))
}
got, err = adp.UserUpdateTags(uid, nil, nil, resetTags)
if err != nil {
t.Fatal(err)
}
want = []string{"alice", "tag111", "tag333"}
if !reflect.DeepEqual(got, want) {
t.Error(mismatchErrorString("Tags", got, want))
}
got, err = adp.UserUpdateTags(uid, addTags, removeTags, nil)
if err != nil {
t.Fatal(err)
}
want = []string{"tag111", "tag333"}
if !reflect.DeepEqual(got, want) {
t.Error(mismatchErrorString("Tags", got, want))
}
got, err = adp.UserUpdateTags(uid, addTags, removeTags, nil)
if err != nil {
t.Fatal(err)
}
want = []string{"tag111", "tag333"}
if !reflect.DeepEqual(got, want) {
t.Error(mismatchErrorString("Tags", got, want))
if got != nil {
t.Fatalf("user tags must stay disabled, got %v", got)
}
}
@@ -1505,6 +1547,11 @@ func init() {
if adp.IsOpen() {
log.Print("Connection is already opened")
}
testData = test_data.InitTestData()
if testData == nil {
log.Fatal("Failed to initialize test data")
}
store.SetTestUidGenerator(*testData.UGen)
err := adp.Open(config.Adapters[adp.GetName()])
if err != nil {
@@ -1512,9 +1559,4 @@ func init() {
}
db = adp.GetTestDB().(*sqlx.DB)
testData = test_data.InitTestData()
if testData == nil {
log.Fatal("Failed to initialize test data")
}
store.SetTestUidGenerator(*testData.UGen)
}
server/db/mysql/user_identity.go
+17
View File
@@ -0,0 +1,17 @@
//go:build mysql
// +build mysql
package mysql
func withoutUserTags(update map[string]any) map[string]any {
if update == nil {
return nil
}
clean := make(map[string]any, len(update))
for key, value := range update {
if key != "Tags" && key != "tags" {
clean[key] = value
}
}
return clean
}
server/db/mysql/user_identity_schema.go
+304
View File
@@ -0,0 +1,304 @@
//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,"
}
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<>''"); err != nil {
return err
}
for _, item := range existing {
seen[identityUniqueKey(item.TenantID, item.Value, scoped)] = true
}
updatedUsers := make(map[string]bool)
for _, item := range candidates {
item.Value = strings.TrimSpace(strings.TrimPrefix(strings.TrimPrefix(item.Value, "tel:"), "basic:"))
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
}