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

fix tenant scoping for group business tables

Browse Source
This commit is contained in:
richjohnrambo
2026-09-23 10:10:08 +08:00
parent 35a70ce9ff
commit 7f0903cd31
6 changed files with 119 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/group_schema.go
+35 -35
View File
@@ -19,9 +19,9 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
version INT NOT NULL DEFAULT 1, created_by BIGINT NOT NULL, dissolved_by BIGINT, 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), 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), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_settings_topic(topic), PRIMARY KEY(id), UNIQUE INDEX uk_im_group_settings_topic(tenant_id, topic),
INDEX idx_im_group_settings_owner(owner_user_id, group_status), INDEX idx_im_group_settings_owner(tenant_id, owner_user_id, group_status),
INDEX idx_im_group_settings_status_updated(group_status, updated_at) INDEX idx_im_group_settings_status_updated(tenant_id, group_status, updated_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_messages ( `CREATE TABLE IF NOT EXISTS im_group_messages (
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, topic VARCHAR(64) NOT NULL, group_id BIGINT NOT NULL, tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, topic VARCHAR(64) NOT NULL, group_id BIGINT NOT NULL,
@@ -53,10 +53,10 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
removed_by BIGINT, version INT NOT NULL DEFAULT 1, removed_by BIGINT, version INT NOT NULL DEFAULT 1,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(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), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_members_topic_user(topic, user_id), PRIMARY KEY(id), UNIQUE INDEX uk_im_group_members_topic_user(tenant_id, topic, user_id),
INDEX idx_im_group_members_user_status(user_id, member_status, updated_at), INDEX idx_im_group_members_user_status(tenant_id, user_id, member_status, updated_at),
INDEX idx_im_group_members_topic_role(topic, role, member_status, id), INDEX idx_im_group_members_topic_role(tenant_id, topic, role, member_status, id),
INDEX idx_im_group_members_topic_status(topic, member_status, id) INDEX idx_im_group_members_topic_status(tenant_id, topic, member_status, id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_invitations ( `CREATE TABLE IF NOT EXISTS im_group_invitations (
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, invitation_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL, tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, invitation_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
@@ -66,10 +66,10 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
expires_at DATETIME(3) NOT NULL, handled_at DATETIME(3), expires_at DATETIME(3) NOT NULL, handled_at DATETIME(3),
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(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), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_invitations_invitation(invitation_id), PRIMARY KEY(id), UNIQUE INDEX uk_im_group_invitations_invitation(tenant_id, invitation_id),
UNIQUE INDEX uk_im_group_invitations_idempotency(inviter_user_id, idempotency_key), UNIQUE INDEX uk_im_group_invitations_idempotency(tenant_id, inviter_user_id, idempotency_key),
INDEX idx_im_group_invitations_target(target_user_id, status, expires_at), INDEX idx_im_group_invitations_target(tenant_id, target_user_id, status, expires_at),
INDEX idx_im_group_invitations_topic_status(topic, status, id) INDEX idx_im_group_invitations_topic_status(tenant_id, topic, status, id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_join_requests ( `CREATE TABLE IF NOT EXISTS im_group_join_requests (
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, request_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL, tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, request_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
@@ -80,11 +80,11 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
idempotency_key VARCHAR(128) NOT NULL DEFAULT '', version INT NOT NULL DEFAULT 1, idempotency_key VARCHAR(128) NOT NULL DEFAULT '', version INT NOT NULL DEFAULT 1,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(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), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_join_requests_request(request_id), PRIMARY KEY(id), UNIQUE INDEX uk_im_group_join_requests_request(tenant_id, request_id),
UNIQUE INDEX uk_im_group_join_requests_idempotency(applicant_user_id, idempotency_key), UNIQUE INDEX uk_im_group_join_requests_idempotency(tenant_id, applicant_user_id, idempotency_key),
INDEX idx_im_group_join_requests_topic_status(topic, status, id), INDEX idx_im_group_join_requests_topic_status(tenant_id, topic, status, id),
INDEX idx_im_group_join_requests_applicant(applicant_user_id, topic, status, id), INDEX idx_im_group_join_requests_applicant(tenant_id, applicant_user_id, topic, status, id),
INDEX idx_im_group_join_requests_expire(status, expires_at) INDEX idx_im_group_join_requests_expire(tenant_id, status, expires_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_announcements ( `CREATE TABLE IF NOT EXISTS im_group_announcements (
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, announcement_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL, tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, announcement_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
@@ -93,17 +93,17 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
idempotency_key VARCHAR(128) NOT NULL DEFAULT '', idempotency_key VARCHAR(128) NOT NULL DEFAULT '',
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(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), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_announcements_announcement(announcement_id), PRIMARY KEY(id), UNIQUE INDEX uk_im_group_announcements_announcement(tenant_id, announcement_id),
UNIQUE INDEX uk_im_group_announcements_idempotency(topic, author_user_id, idempotency_key), UNIQUE INDEX uk_im_group_announcements_idempotency(tenant_id, topic, author_user_id, idempotency_key),
INDEX idx_im_group_announcements_topic_status(topic, status, version DESC, id DESC), INDEX idx_im_group_announcements_topic_status(tenant_id, topic, status, version DESC, id DESC),
INDEX idx_im_group_announcements_topic_time(topic, published_at DESC, id DESC) INDEX idx_im_group_announcements_topic_time(tenant_id, topic, published_at DESC, id DESC)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_announcement_reads ( `CREATE TABLE IF NOT EXISTS im_group_announcement_reads (
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL, announcement_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, tenant_id BIGINT NOT NULL DEFAULT 0, 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), 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), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_announcement_reads_announcement_user(announcement_id, user_id), PRIMARY KEY(id), UNIQUE INDEX uk_im_group_announcement_reads_announcement_user(tenant_id, announcement_id, user_id),
INDEX idx_im_group_announcement_reads_topic_user(topic, user_id, read_at) INDEX idx_im_group_announcement_reads_topic_user(tenant_id, topic, user_id, read_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_mutes ( `CREATE TABLE IF NOT EXISTS im_group_mutes (
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL, topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, muted_by BIGINT NOT NULL, tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL, topic VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, muted_by BIGINT NOT NULL,
@@ -111,9 +111,9 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
status VARCHAR(16) NOT NULL DEFAULT 'active', revoked_by BIGINT, revoked_at DATETIME(3), 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), 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), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_mutes_topic_user(topic, user_id), PRIMARY KEY(id), UNIQUE INDEX uk_im_group_mutes_topic_user(tenant_id, topic, user_id),
INDEX idx_im_group_mutes_topic_status(topic, status, expires_at), INDEX idx_im_group_mutes_topic_status(tenant_id, topic, status, expires_at),
INDEX idx_im_group_mutes_user_status(user_id, status, expires_at) INDEX idx_im_group_mutes_user_status(tenant_id, user_id, status, expires_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_qrcodes ( `CREATE TABLE IF NOT EXISTS im_group_qrcodes (
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, token_id VARCHAR(64) NOT NULL, token_hash CHAR(64) NOT NULL, tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, token_id VARCHAR(64) NOT NULL, token_hash CHAR(64) NOT NULL,
@@ -121,8 +121,8 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
expires_at DATETIME(3) NOT NULL, revoked_by BIGINT, revoked_at DATETIME(3), expires_at DATETIME(3) NOT NULL, revoked_by BIGINT, revoked_at DATETIME(3),
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(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), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_qrcodes_token(token_id), PRIMARY KEY(id), UNIQUE INDEX uk_im_group_qrcodes_token(tenant_id, token_id),
UNIQUE INDEX uk_im_group_qrcodes_hash(token_hash), INDEX idx_im_group_qrcodes_topic_status(topic, status, expires_at) UNIQUE INDEX uk_im_group_qrcodes_hash(tenant_id, token_hash), INDEX idx_im_group_qrcodes_topic_status(tenant_id, topic, status, expires_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_files ( `CREATE TABLE IF NOT EXISTS im_group_files (
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, group_file_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL, tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, group_file_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
@@ -133,11 +133,11 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
idempotency_key VARCHAR(128) NOT NULL DEFAULT '', idempotency_key VARCHAR(128) NOT NULL DEFAULT '',
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(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), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_files_group_file(group_file_id), PRIMARY KEY(id), UNIQUE INDEX uk_im_group_files_group_file(tenant_id, group_file_id),
UNIQUE INDEX uk_im_group_files_topic_file(topic, file_id), UNIQUE INDEX uk_im_group_files_topic_file(tenant_id, topic, file_id),
UNIQUE INDEX uk_im_group_files_idempotency(topic, uploader_id, idempotency_key), UNIQUE INDEX uk_im_group_files_idempotency(tenant_id, topic, uploader_id, idempotency_key),
INDEX idx_im_group_files_topic_type(topic, status, file_type, id DESC), INDEX idx_im_group_files_topic_type(tenant_id, topic, status, file_type, id DESC),
INDEX idx_im_group_files_topic_name(topic, status, file_name), INDEX idx_im_group_files_uploader(uploader_id, id DESC) INDEX idx_im_group_files_topic_name(tenant_id, topic, status, file_name), INDEX idx_im_group_files_uploader(tenant_id, uploader_id, id DESC)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_group_admin_logs ( `CREATE TABLE IF NOT EXISTS im_group_admin_logs (
tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, log_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL, tenant_id BIGINT NOT NULL DEFAULT 0, id BIGINT NOT NULL AUTO_INCREMENT, log_id VARCHAR(64) NOT NULL, topic VARCHAR(32) NOT NULL,
@@ -145,9 +145,9 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
object_type VARCHAR(32) NOT NULL DEFAULT '', object_id VARCHAR(64) NOT NULL DEFAULT '', object_type VARCHAR(32) NOT NULL DEFAULT '', object_id VARCHAR(64) NOT NULL DEFAULT '',
before_data JSON, after_data JSON, request_id VARCHAR(128) NOT NULL DEFAULT '', before_data JSON, after_data JSON, request_id VARCHAR(128) NOT NULL DEFAULT '',
idempotency_key VARCHAR(128) NOT NULL DEFAULT '', created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), idempotency_key VARCHAR(128) NOT NULL DEFAULT '', created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_admin_logs_log(log_id), PRIMARY KEY(id), UNIQUE INDEX uk_im_group_admin_logs_log(tenant_id, log_id),
INDEX idx_im_group_admin_logs_topic_time(topic, id DESC), INDEX idx_im_group_admin_logs_topic_time(tenant_id, topic, id DESC),
INDEX idx_im_group_admin_logs_operator(operator_id, id DESC), INDEX idx_im_group_admin_logs_object(object_type, object_id) INDEX idx_im_group_admin_logs_operator(tenant_id, operator_id, id DESC), INDEX idx_im_group_admin_logs_object(tenant_id, object_type, object_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`UPDATE im_topics SET public=JSON_SET(public, '$.fn', JSON_UNQUOTE(JSON_EXTRACT(public, '$.name'))) `UPDATE im_topics SET public=JSON_SET(public, '$.fn', JSON_UNQUOTE(JSON_EXTRACT(public, '$.name')))
WHERE name LIKE 'grp%' AND JSON_VALID(public) WHERE name LIKE 'grp%' AND JSON_VALID(public)
server/db/mysql/group_schema_test.go
+3 -3
View File
@@ -40,9 +40,9 @@ func TestEnsureIMGroupBusinessTablesUsesSelectedTenantDatabase(t *testing.T) {
t.Fatal("group idempotency table must not be created") t.Fatal("group idempotency table must not be created")
} }
for _, expected := range []string{ for _, expected := range []string{
"primary key(id), unique index uk_im_group_settings_topic(topic)", "primary key(id), unique index uk_im_group_settings_topic(tenant_id, 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_announcement_reads_announcement_user(tenant_id, announcement_id, user_id)",
"primary key(id), unique index uk_im_group_mutes_topic_user(topic, user_id)", "primary key(id), unique index uk_im_group_mutes_topic_user(tenant_id, topic, user_id)",
} { } {
if !strings.Contains(joined, expected) { if !strings.Contains(joined, expected) {
t.Fatalf("group schema is missing snowflake primary-key definition %q", expected) t.Fatalf("group schema is missing snowflake primary-key definition %q", expected)
server/db/mysql/snowflake_primary_keys.go
+3 -3
View File
@@ -29,9 +29,9 @@ var snowflakePrimaryKeySpecs = []snowflakePrimaryKeySpec{
{table: "im_friendships", uniqueIndex: "uk_im_friendships_pair", businessKey: []string{"user_id", "peer_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_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_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_settings", uniqueIndex: "uk_im_group_settings_topic", businessKey: []string{"tenant_id", "topic"}},
{table: "im_group_announcement_reads", uniqueIndex: "uk_im_group_announcement_reads_announcement_user", businessKey: []string{"announcement_id", "user_id"}}, {table: "im_group_announcement_reads", uniqueIndex: "uk_im_group_announcement_reads_announcement_user", businessKey: []string{"tenant_id", "announcement_id", "user_id"}},
{table: "im_group_mutes", uniqueIndex: "uk_im_group_mutes_topic_user", businessKey: []string{"topic", "user_id"}}, {table: "im_group_mutes", uniqueIndex: "uk_im_group_mutes_topic_user", businessKey: []string{"tenant_id", "topic", "user_id"}},
} }
const snowflakeMigrationLock = "CONCAT('im_snowflake_pk:', LEFT(SHA2(DATABASE(), 256), 40))" const snowflakeMigrationLock = "CONCAT('im_snowflake_pk:', LEFT(SHA2(DATABASE(), 256), 40))"
server/db/mysql/tenant_adapter.go
+1 -1
View File
@@ -5270,7 +5270,7 @@ func (a *adapter) TenantFileUserHasAccess(id t.TenantID, fid string, uid t.Uid)
SELECT 1 FROM im_group_files AS gf SELECT 1 FROM im_group_files AS gf
INNER JOIN im_subscriptions AS s INNER JOIN im_subscriptions AS s
ON s.tenant_id=fu.tenant_id AND s.topic=gf.topic AND s.userid=? ON s.tenant_id=fu.tenant_id AND s.topic=gf.topic AND s.userid=?
WHERE (gf.file_id=? OR gf.file_id LIKE CONCAT(?,'.%%')) AND gf.status='active' WHERE gf.tenant_id=fu.tenant_id AND (gf.file_id=? OR gf.file_id LIKE CONCAT(?,'.%%')) AND gf.status='active'
AND s.deletedat IS NULL AND INSTR(s.modewant,'R')>0 AND INSTR(s.modegiven,'R')>0 AND s.deletedat IS NULL AND INSTR(s.modewant,'R')>0 AND INSTR(s.modegiven,'R')>0
) )
) LIMIT 1`, tenantFriendshipsTable(id), tenantMessagesTable(id))), ) LIMIT 1`, tenantFriendshipsTable(id), tenantMessagesTable(id))),
server/db/mysql/tenant_groups.go
+72 -72
View File
@@ -20,12 +20,12 @@ const (
groupSettingsSelectColumns = `tenant_id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all, groupSettingsSelectColumns = `tenant_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, 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` created_by,dissolved_by,dissolved_at,created_at,updated_at`
groupAnnouncementSelectColumnsBase = `id,announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at, groupAnnouncementSelectColumnsBase = `tenant_id,id,announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at,
(SELECT COUNT(*) FROM im_group_announcement_reads AS ar WHERE ar.topic=im_group_announcements.topic AND ar.announcement_id=im_group_announcements.announcement_id) AS read_count, (SELECT COUNT(*) FROM im_group_announcement_reads AS ar WHERE ar.tenant_id=im_group_announcements.tenant_id AND ar.topic=im_group_announcements.topic AND ar.announcement_id=im_group_announcements.announcement_id) AS read_count,
(SELECT COUNT(*) FROM im_group_members AS gm WHERE gm.topic=im_group_announcements.topic AND gm.member_status='active') AS member_count` (SELECT COUNT(*) FROM im_group_members AS gm WHERE gm.tenant_id=im_group_announcements.tenant_id AND gm.topic=im_group_announcements.topic AND gm.member_status='active') AS member_count`
groupMemberSelectColumns = `id,topic,user_id,role,member_status,nickname,pinned,mute_notifications, groupMemberSelectColumns = `id,topic,user_id,role,member_status,nickname,pinned,mute_notifications,
join_source,joined_by,joined_at,left_at,removed_by,version,created_at,updated_at` join_source,joined_by,joined_at,left_at,removed_by,version,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` groupMuteSelectColumns = `tenant_id,topic,user_id,muted_by,muted_at,expires_at,reason,status,revoked_by,revoked_at,version,created_at,updated_at`
) )
func groupAnnouncementSelectColumns(tenantID t.TenantID) string { func groupAnnouncementSelectColumns(tenantID t.TenantID) string {
@@ -488,13 +488,13 @@ func (a *adapter) TenantGroupInvitationCreate(id t.TenantID, inv *t.GroupInvitat
defer cancel() defer cancel()
} }
if strings.TrimSpace(inv.JoinRequestID) != "" { if strings.TrimSpace(inv.JoinRequestID) != "" {
existing, err := a.groupInvitationByQuery(ctx, `SELECT inv.id,inv.invitation_id,inv.topic,inv.inviter_user_id,inv.target_user_id,inv.message,inv.status,inv.result, existing, err := a.groupInvitationByQuery(ctx, `SELECT inv.tenant_id,inv.id,inv.invitation_id,inv.topic,inv.inviter_user_id,inv.target_user_id,inv.message,inv.status,inv.result,
inv.join_request_id,inv.idempotency_key,inv.expires_at,inv.handled_at,inv.created_at,inv.updated_at inv.join_request_id,inv.idempotency_key,inv.expires_at,inv.handled_at,inv.created_at,inv.updated_at
FROM im_group_invitations AS inv FROM im_group_invitations AS inv
INNER JOIN im_group_join_requests AS req ON req.request_id=inv.join_request_id INNER JOIN im_group_join_requests AS req ON req.tenant_id=inv.tenant_id AND req.request_id=inv.join_request_id
WHERE inv.topic=? AND inv.target_user_id=? AND inv.status=? AND req.status=? WHERE inv.tenant_id=? AND inv.topic=? AND inv.target_user_id=? AND inv.status=? AND req.status=?
ORDER BY req.id DESC LIMIT 1`, ORDER BY req.id DESC LIMIT 1`,
inv.Topic, decodeUID(inv.TargetUserID), t.GroupInvitationStatusPending, t.GroupJoinRequestStatusPending) id, inv.Topic, decodeUID(inv.TargetUserID), t.GroupInvitationStatusPending, t.GroupJoinRequestStatusPending)
if err == nil { if err == nil {
return existing, nil return existing, nil
} }
@@ -507,16 +507,16 @@ func (a *adapter) TenantGroupInvitationCreate(id t.TenantID, inv *t.GroupInvitat
return nil, err return nil, err
} }
defer tx.Rollback() defer tx.Rollback()
_, err = tx.ExecContext(ctx, `INSERT INTO im_group_invitations(invitation_id,topic,inviter_user_id,target_user_id,message,status,result,join_request_id,idempotency_key,expires_at,handled_at,created_at,updated_at) _, err = tx.ExecContext(ctx, `INSERT INTO im_group_invitations(tenant_id,invitation_id,topic,inviter_user_id,target_user_id,message,status,result,join_request_id,idempotency_key,expires_at,handled_at,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
inv.InvitationID, inv.Topic, decodeUID(inv.InviterUserID), decodeUID(inv.TargetUserID), inv.Message, inv.Status, inv.Result, inv.JoinRequestID, inv.IdempotencyKey, inv.ExpiresAt, inv.HandledAt, now, now) id, inv.InvitationID, inv.Topic, decodeUID(inv.InviterUserID), decodeUID(inv.TargetUserID), inv.Message, inv.Status, inv.Result, inv.JoinRequestID, inv.IdempotencyKey, inv.ExpiresAt, inv.HandledAt, now, now)
if err != nil { if err != nil {
return nil, err return nil, err
} }
if strings.TrimSpace(inv.JoinRequestID) != "" { if strings.TrimSpace(inv.JoinRequestID) != "" {
_, err = tx.ExecContext(ctx, `INSERT INTO im_group_join_requests(request_id,topic,applicant_user_id,inviter_user_id,source,qr_token_hash,message,status,reject_reason,expires_at,idempotency_key,version,created_at,updated_at) _, err = tx.ExecContext(ctx, `INSERT INTO im_group_join_requests(tenant_id,request_id,topic,applicant_user_id,inviter_user_id,source,qr_token_hash,message,status,reject_reason,expires_at,idempotency_key,version,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
inv.JoinRequestID, inv.Topic, decodeUID(inv.TargetUserID), nullableUID(inv.InviterUserID), "invitation", "", inv.Message, t.GroupJoinRequestStatusPending, "", inv.ExpiresAt, inv.IdempotencyKey, 1, now, now) id, inv.JoinRequestID, inv.Topic, decodeUID(inv.TargetUserID), nullableUID(inv.InviterUserID), "invitation", "", inv.Message, t.GroupJoinRequestStatusPending, "", inv.ExpiresAt, inv.IdempotencyKey, 1, now, now)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -548,10 +548,10 @@ func (a *adapter) TenantGroupInvitationAccept(id t.TenantID, invitationID string
var inv t.GroupInvitation var inv t.GroupInvitation
var handledAt sql.NullTime var handledAt sql.NullTime
err = tx.QueryRowxContext(ctx, `SELECT id,invitation_id,topic,inviter_user_id,target_user_id,message,status,result, err = tx.QueryRowxContext(ctx, `SELECT tenant_id,id,invitation_id,topic,inviter_user_id,target_user_id,message,status,result,
join_request_id,idempotency_key,expires_at,handled_at,created_at,updated_at join_request_id,idempotency_key,expires_at,handled_at,created_at,updated_at
FROM im_group_invitations WHERE invitation_id=? FOR UPDATE`, strings.TrimSpace(invitationID)). FROM im_group_invitations WHERE tenant_id=? AND invitation_id=? FOR UPDATE`, id, strings.TrimSpace(invitationID)).
Scan(&inv.ID, &inv.InvitationID, &inv.Topic, &inv.InviterUserRaw, &inv.TargetUserRaw, &inv.Message, &inv.Status, Scan(&inv.TenantID, &inv.ID, &inv.InvitationID, &inv.Topic, &inv.InviterUserRaw, &inv.TargetUserRaw, &inv.Message, &inv.Status,
&inv.Result, &inv.JoinRequestID, &inv.IdempotencyKey, &inv.ExpiresAt, &handledAt, &inv.CreatedAt, &inv.UpdatedAt) &inv.Result, &inv.JoinRequestID, &inv.IdempotencyKey, &inv.ExpiresAt, &handledAt, &inv.CreatedAt, &inv.UpdatedAt)
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
return nil, t.ErrNotFound return nil, t.ErrNotFound
@@ -606,15 +606,15 @@ func (a *adapter) TenantGroupInvitationAccept(id t.TenantID, invitationID string
return nil, err return nil, err
} }
if _, err = tx.ExecContext(ctx, `UPDATE im_group_invitations if _, err = tx.ExecContext(ctx, `UPDATE im_group_invitations
SET status=?,result=?,handled_at=?,updated_at=? WHERE invitation_id=?`, SET status=?,result=?,handled_at=?,updated_at=? WHERE tenant_id=? AND invitation_id=?`,
t.GroupInvitationStatusJoined, "joined", now, now, inv.InvitationID); err != nil { t.GroupInvitationStatusJoined, "joined", now, now, id, inv.InvitationID); err != nil {
return nil, err return nil, err
} }
if strings.TrimSpace(inv.JoinRequestID) != "" { if strings.TrimSpace(inv.JoinRequestID) != "" {
if _, err = tx.ExecContext(ctx, `UPDATE im_group_join_requests if _, err = tx.ExecContext(ctx, `UPDATE im_group_join_requests
SET status=?,handled_by=?,handled_at=?,updated_at=?,version=version+1 SET status=?,handled_by=?,handled_at=?,updated_at=?,version=version+1
WHERE topic=? AND request_id=? AND status=?`, WHERE tenant_id=? AND topic=? AND request_id=? AND status=?`,
t.GroupJoinRequestStatusAccepted, decodeUID(uid), now, now, inv.Topic, inv.JoinRequestID, t.GroupJoinRequestStatusPending); err != nil { t.GroupJoinRequestStatusAccepted, decodeUID(uid), now, now, id, inv.Topic, inv.JoinRequestID, t.GroupJoinRequestStatusPending); err != nil {
return nil, err return nil, err
} }
} }
@@ -651,8 +651,8 @@ func (a *adapter) TenantGroupJoinRequestCreate(id t.TenantID, req *t.GroupJoinRe
if cancel != nil { if cancel != nil {
defer cancel() defer cancel()
} }
existing, err := a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE topic=? AND applicant_user_id=? AND status=? ORDER BY id DESC LIMIT 1", existing, err := a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE tenant_id=? AND topic=? AND applicant_user_id=? AND status=? ORDER BY id DESC LIMIT 1",
req.Topic, decodeUID(req.ApplicantUserID), t.GroupJoinRequestStatusPending) id, req.Topic, decodeUID(req.ApplicantUserID), t.GroupJoinRequestStatusPending)
if err == nil { if err == nil {
return existing, nil return existing, nil
} }
@@ -664,16 +664,16 @@ func (a *adapter) TenantGroupJoinRequestCreate(id t.TenantID, req *t.GroupJoinRe
return nil, err return nil, err
} }
defer tx.Rollback() defer tx.Rollback()
_, err = tx.ExecContext(ctx, `INSERT INTO im_group_join_requests(request_id,topic,applicant_user_id,inviter_user_id,source,qr_token_hash,message,status,reject_reason,expires_at,idempotency_key,version,created_at,updated_at) _, err = tx.ExecContext(ctx, `INSERT INTO im_group_join_requests(tenant_id,request_id,topic,applicant_user_id,inviter_user_id,source,qr_token_hash,message,status,reject_reason,expires_at,idempotency_key,version,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
req.RequestID, req.Topic, decodeUID(req.ApplicantUserID), nullableUID(req.InviterUserID), req.Source, req.QRTokenHash, req.Message, req.Status, req.RejectReason, req.ExpiresAt, req.IdempotencyKey, req.Version, now, now) id, req.RequestID, req.Topic, decodeUID(req.ApplicantUserID), nullableUID(req.InviterUserID), req.Source, req.QRTokenHash, req.Message, req.Status, req.RejectReason, req.ExpiresAt, req.IdempotencyKey, req.Version, now, now)
if err != nil { if err != nil {
return nil, err return nil, err
} }
if err = tx.Commit(); err != nil { if err = tx.Commit(); err != nil {
return nil, err return nil, err
} }
return a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE request_id=?", req.RequestID) return a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE tenant_id=? AND request_id=?", id, req.RequestID)
} }
func (a *adapter) TenantGroupJoinRequestList(id t.TenantID, topic, status, cursor string, limit int) ([]t.GroupJoinRequest, string, error) { func (a *adapter) TenantGroupJoinRequestList(id t.TenantID, topic, status, cursor string, limit int) ([]t.GroupJoinRequest, string, error) {
@@ -682,15 +682,15 @@ func (a *adapter) TenantGroupJoinRequestList(id t.TenantID, topic, status, curso
} else if tenantAdp != a { } else if tenantAdp != a {
return tenantAdp.TenantGroupJoinRequestList(id, topic, status, cursor, limit) return tenantAdp.TenantGroupJoinRequestList(id, topic, status, cursor, limit)
} }
where := []string{"r.topic=?"} where := []string{"r.tenant_id=?", "r.topic=?"}
args := []any{topic} args := []any{id, topic}
if status != "" && status != "all" { if status != "" && status != "all" {
where = append(where, "r.status=?") where = append(where, "r.status=?")
args = append(args, status) args = append(args, status)
if status == t.GroupJoinRequestStatusPending { if status == t.GroupJoinRequestStatusPending {
where = append(where, `NOT EXISTS ( where = append(where, `NOT EXISTS (
SELECT 1 FROM im_group_join_requests AS newer SELECT 1 FROM im_group_join_requests AS newer
WHERE newer.topic=r.topic AND newer.applicant_user_id=r.applicant_user_id WHERE newer.tenant_id=r.tenant_id AND newer.topic=r.topic AND newer.applicant_user_id=r.applicant_user_id
AND newer.status=? AND newer.id>r.id AND newer.status=? AND newer.id>r.id
)`) )`)
args = append(args, status) args = append(args, status)
@@ -751,7 +751,7 @@ func (a *adapter) TenantGroupJoinRequestResolve(id t.TenantID, topic, requestID
} }
defer tx.Rollback() defer tx.Rollback()
var current t.GroupJoinRequest var current t.GroupJoinRequest
err = tx.QueryRowxContext(ctx, "SELECT * FROM im_group_join_requests WHERE topic=? AND request_id=? FOR UPDATE", topic, requestID).StructScan(&current) err = tx.QueryRowxContext(ctx, "SELECT * FROM im_group_join_requests WHERE tenant_id=? AND topic=? AND request_id=? FOR UPDATE", id, topic, requestID).StructScan(&current)
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
return nil, t.ErrNotFound return nil, t.ErrNotFound
} }
@@ -780,7 +780,7 @@ func (a *adapter) TenantGroupJoinRequestResolve(id t.TenantID, topic, requestID
} }
} }
if _, err = tx.ExecContext(ctx, `UPDATE im_group_join_requests SET status=?,reject_reason=?,handled_by=?,handled_at=?,updated_at=?,version=version+1 if _, err = tx.ExecContext(ctx, `UPDATE im_group_join_requests SET status=?,reject_reason=?,handled_by=?,handled_at=?,updated_at=?,version=version+1
WHERE topic=? AND request_id=? AND status=?`, status, rejectReason, decodeUID(actor), now, now, topic, requestID, t.GroupJoinRequestStatusPending); err != nil { WHERE tenant_id=? AND topic=? AND request_id=? AND status=?`, status, rejectReason, decodeUID(actor), now, now, id, topic, requestID, t.GroupJoinRequestStatusPending); err != nil {
return nil, err return nil, err
} }
if action == "accept" { if action == "accept" {
@@ -794,7 +794,7 @@ func (a *adapter) TenantGroupJoinRequestResolve(id t.TenantID, topic, requestID
if err = tx.Commit(); err != nil { if err = tx.Commit(); err != nil {
return nil, err return nil, err
} }
return a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE topic=? AND request_id=?", topic, requestID) return a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE tenant_id=? AND topic=? AND request_id=?", id, topic, requestID)
} }
func (a *adapter) TenantGroupAnnouncementLatest(id t.TenantID, topic string) (*t.GroupAnnouncement, error) { func (a *adapter) TenantGroupAnnouncementLatest(id t.TenantID, topic string) (*t.GroupAnnouncement, error) {
@@ -807,7 +807,7 @@ func (a *adapter) TenantGroupAnnouncementLatest(id t.TenantID, topic string) (*t
if cancel != nil { if cancel != nil {
defer cancel() defer cancel()
} }
return a.groupAnnouncementByQuery(ctx, "SELECT "+groupAnnouncementSelectColumns(id)+" FROM im_group_announcements WHERE topic=? AND status=? ORDER BY version DESC,id DESC LIMIT 1", topic, t.GroupAnnouncementStatusCurrent) return a.groupAnnouncementByQuery(ctx, "SELECT "+groupAnnouncementSelectColumns(id)+" FROM im_group_announcements WHERE tenant_id=? AND topic=? AND status=? ORDER BY version DESC,id DESC LIMIT 1", id, topic, t.GroupAnnouncementStatusCurrent)
} }
func (a *adapter) TenantGroupAnnouncementGet(id t.TenantID, topic, announcementID string) (*t.GroupAnnouncement, error) { func (a *adapter) TenantGroupAnnouncementGet(id t.TenantID, topic, announcementID string) (*t.GroupAnnouncement, error) {
@@ -820,7 +820,7 @@ func (a *adapter) TenantGroupAnnouncementGet(id t.TenantID, topic, announcementI
if cancel != nil { if cancel != nil {
defer cancel() defer cancel()
} }
return a.groupAnnouncementByQuery(ctx, "SELECT "+groupAnnouncementSelectColumns(id)+" FROM im_group_announcements WHERE topic=? AND announcement_id=? LIMIT 1", topic, announcementID) return a.groupAnnouncementByQuery(ctx, "SELECT "+groupAnnouncementSelectColumns(id)+" FROM im_group_announcements WHERE tenant_id=? AND topic=? AND announcement_id=? LIMIT 1", id, topic, announcementID)
} }
func (a *adapter) TenantGroupAnnouncementPut(id t.TenantID, ann *t.GroupAnnouncement, expectedVersion int) (*t.GroupAnnouncement, error) { func (a *adapter) TenantGroupAnnouncementPut(id t.TenantID, ann *t.GroupAnnouncement, expectedVersion int) (*t.GroupAnnouncement, error) {
@@ -843,7 +843,7 @@ func (a *adapter) TenantGroupAnnouncementPut(id t.TenantID, ann *t.GroupAnnounce
} }
}() }()
var currentVersion int var currentVersion int
err = tx.QueryRowContext(ctx, "SELECT COALESCE(MAX(version),0) FROM im_group_announcements WHERE topic=?", ann.Topic).Scan(&currentVersion) err = tx.QueryRowContext(ctx, "SELECT COALESCE(MAX(version),0) FROM im_group_announcements WHERE tenant_id=? AND topic=?", id, ann.Topic).Scan(&currentVersion)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -855,7 +855,7 @@ func (a *adapter) TenantGroupAnnouncementPut(id t.TenantID, ann *t.GroupAnnounce
if now.IsZero() { if now.IsZero() {
now = time.Now().UTC().Round(time.Millisecond) now = time.Now().UTC().Round(time.Millisecond)
} }
if _, err = tx.ExecContext(ctx, "UPDATE im_group_announcements SET status=?,updated_at=? WHERE topic=? AND status=?", t.GroupAnnouncementStatusSuperseded, now, ann.Topic, t.GroupAnnouncementStatusCurrent); err != nil { if _, err = tx.ExecContext(ctx, "UPDATE im_group_announcements SET status=?,updated_at=? WHERE tenant_id=? AND topic=? AND status=?", t.GroupAnnouncementStatusSuperseded, now, id, ann.Topic, t.GroupAnnouncementStatusCurrent); err != nil {
return nil, err return nil, err
} }
if ann.AnnouncementID == "" { if ann.AnnouncementID == "" {
@@ -864,8 +864,8 @@ func (a *adapter) TenantGroupAnnouncementPut(id t.TenantID, ann *t.GroupAnnounce
if ann.IdempotencyKey == "" { if ann.IdempotencyKey == "" {
ann.IdempotencyKey = ann.AnnouncementID ann.IdempotencyKey = ann.AnnouncementID
} }
if _, err = tx.ExecContext(ctx, `INSERT INTO im_group_announcements(announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at) if _, err = tx.ExecContext(ctx, `INSERT INTO im_group_announcements(tenant_id,announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?)`, ann.AnnouncementID, ann.Topic, decodeUID(ann.AuthorUserID), ann.Content, t.GroupAnnouncementStatusCurrent, version, now, ann.IdempotencyKey, now, now); err != nil { VALUES(?,?,?,?,?,?,?,?,?,?,?)`, id, ann.AnnouncementID, ann.Topic, decodeUID(ann.AuthorUserID), ann.Content, t.GroupAnnouncementStatusCurrent, version, now, ann.IdempotencyKey, now, now); err != nil {
return nil, err return nil, err
} }
if err = tx.Commit(); err != nil { if err = tx.Commit(); err != nil {
@@ -888,8 +888,8 @@ func (a *adapter) TenantGroupAnnouncementRead(id t.TenantID, topic, announcement
if err != nil { if err != nil {
return err return err
} }
_, err = a.db.ExecContext(ctx, `INSERT INTO im_group_announcement_reads(id,announcement_id,topic,user_id,read_at) _, err = a.db.ExecContext(ctx, `INSERT INTO im_group_announcement_reads(id,tenant_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) VALUES(?,?,?,?,?,?) ON DUPLICATE KEY UPDATE read_at=VALUES(read_at),updated_at=CURRENT_TIMESTAMP(3)`, readID, id, announcementID, topic, decodeUID(uid), readAt)
return err return err
} }
@@ -899,8 +899,8 @@ func (a *adapter) TenantGroupMuteList(id t.TenantID, topic, cursor string, limit
} else if tenantAdp != a { } else if tenantAdp != a {
return tenantAdp.TenantGroupMuteList(id, topic, cursor, limit) return tenantAdp.TenantGroupMuteList(id, topic, cursor, limit)
} }
where := []string{"topic=?", "status=?"} where := []string{"tenant_id=?", "topic=?", "status=?"}
args := []any{topic, t.GroupMuteStatusActive} args := []any{id, topic, t.GroupMuteStatusActive}
if cursor != "" { if cursor != "" {
parsed, err := strconv.ParseInt(cursor, 10, 64) parsed, err := strconv.ParseInt(cursor, 10, 64)
if err != nil || parsed <= 0 { if err != nil || parsed <= 0 {
@@ -950,8 +950,8 @@ func (a *adapter) TenantGroupMuteGet(id t.TenantID, topic string, uid t.Uid) (*t
if cancel != nil { if cancel != nil {
defer cancel() defer cancel()
} }
return a.groupMuteByQuery(ctx, "SELECT "+groupMuteSelectColumns+" FROM im_group_mutes WHERE topic=? AND user_id=? AND status=? AND (expires_at IS NULL OR expires_at>?)", return a.groupMuteByQuery(ctx, "SELECT "+groupMuteSelectColumns+" FROM im_group_mutes WHERE tenant_id=? AND topic=? AND user_id=? AND status=? AND (expires_at IS NULL OR expires_at>?)",
topic, decodeUID(uid), t.GroupMuteStatusActive, now) id, topic, decodeUID(uid), t.GroupMuteStatusActive, now)
} }
func (a *adapter) TenantGroupMuteUpsert(id t.TenantID, mute *t.GroupMute) (*t.GroupMute, error) { func (a *adapter) TenantGroupMuteUpsert(id t.TenantID, mute *t.GroupMute) (*t.GroupMute, error) {
@@ -980,17 +980,17 @@ func (a *adapter) TenantGroupMuteUpsert(id t.TenantID, mute *t.GroupMute) (*t.Gr
if err != nil { if err != nil {
return nil, err 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) _, err = tx.ExecContext(ctx, `INSERT INTO im_group_mutes(id,tenant_id,topic,user_id,muted_by,muted_at,expires_at,reason,status,version)
VALUES(?,?,?,?,?,?,?,?,1) 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)`, 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)`,
muteID, mute.Topic, decodeUID(mute.UserID), decodeUID(mute.MutedBy), now, mute.ExpiresAt, mute.Reason, mute.Status) muteID, id, mute.Topic, decodeUID(mute.UserID), decodeUID(mute.MutedBy), now, mute.ExpiresAt, mute.Reason, mute.Status)
if err != nil { if err != nil {
return nil, err return nil, err
} }
if err = tx.Commit(); err != nil { if err = tx.Commit(); err != nil {
return nil, err return nil, err
} }
return a.groupMuteByQuery(ctx, "SELECT "+groupMuteSelectColumns+" 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 tenant_id=? AND topic=? AND user_id=?", id, mute.Topic, decodeUID(mute.UserID))
} }
func (a *adapter) TenantGroupMuteDelete(id t.TenantID, topic string, uid, actor t.Uid, now time.Time) error { func (a *adapter) TenantGroupMuteDelete(id t.TenantID, topic string, uid, actor t.Uid, now time.Time) error {
@@ -1008,8 +1008,8 @@ func (a *adapter) TenantGroupMuteDelete(id t.TenantID, topic string, uid, actor
return err return err
} }
defer tx.Rollback() defer tx.Rollback()
res, err := tx.ExecContext(ctx, "UPDATE im_group_mutes SET status=?,revoked_by=?,revoked_at=?,version=version+1,updated_at=? WHERE topic=? AND user_id=? AND status=?", res, err := tx.ExecContext(ctx, "UPDATE im_group_mutes SET status=?,revoked_by=?,revoked_at=?,version=version+1,updated_at=? WHERE tenant_id=? AND topic=? AND user_id=? AND status=?",
t.GroupMuteStatusRevoked, decodeUID(actor), now, now, topic, decodeUID(uid), t.GroupMuteStatusActive) t.GroupMuteStatusRevoked, decodeUID(actor), now, now, id, topic, decodeUID(uid), t.GroupMuteStatusActive)
if err != nil { if err != nil {
return err return err
} }
@@ -1028,8 +1028,8 @@ func (a *adapter) TenantGroupFileList(id t.TenantID, topic, fileType, query, cur
if strings.TrimSpace(topic) == "" { if strings.TrimSpace(topic) == "" {
return nil, "", t.ErrMalformed return nil, "", t.ErrMalformed
} }
where := []string{"topic=?", "status='active'"} where := []string{"tenant_id=?", "topic=?", "status='active'"}
args := []any{strings.TrimSpace(topic)} args := []any{id, strings.TrimSpace(topic)}
if fileType != "" { if fileType != "" {
where = append(where, "file_type=?") where = append(where, "file_type=?")
args = append(args, strings.TrimSpace(fileType)) args = append(args, strings.TrimSpace(fileType))
@@ -1051,7 +1051,7 @@ func (a *adapter) TenantGroupFileList(id t.TenantID, topic, fileType, query, cur
if cancel != nil { if cancel != nil {
defer cancel() defer cancel()
} }
rows, err := a.db.QueryxContext(ctx, `SELECT id,group_file_id,topic,file_id,source_topic,source_seq,uploader_id, rows, err := a.db.QueryxContext(ctx, `SELECT tenant_id,id,group_file_id,topic,file_id,source_topic,source_seq,uploader_id,
file_name,mime_type,file_type,file_size,status,deleted_by,deleted_at,idempotency_key,created_at,updated_at file_name,mime_type,file_type,file_size,status,deleted_by,deleted_at,idempotency_key,created_at,updated_at
FROM im_group_files WHERE `+strings.Join(where, " AND ")+` ORDER BY id DESC LIMIT ?`, args...) FROM im_group_files WHERE `+strings.Join(where, " AND ")+` ORDER BY id DESC LIMIT ?`, args...)
if err != nil { if err != nil {
@@ -1108,15 +1108,15 @@ func (a *adapter) TenantGroupFileCreate(id t.TenantID, file *t.GroupFile) (*t.Gr
} }
defer tx.Rollback() defer tx.Rollback()
_, err = tx.ExecContext(ctx, `INSERT INTO im_group_files _, err = tx.ExecContext(ctx, `INSERT INTO im_group_files
(group_file_id,topic,file_id,source_topic,source_seq,uploader_id,file_name,mime_type,file_type,file_size,status,idempotency_key,created_at,updated_at) (tenant_id,group_file_id,topic,file_id,source_topic,source_seq,uploader_id,file_name,mime_type,file_type,file_size,status,idempotency_key,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE ON DUPLICATE KEY UPDATE
file_name=IF(file_name='', VALUES(file_name), file_name), file_name=IF(file_name='', VALUES(file_name), file_name),
mime_type=IF(mime_type='', VALUES(mime_type), mime_type), mime_type=IF(mime_type='', VALUES(mime_type), mime_type),
file_size=IF(file_size=0, VALUES(file_size), file_size), file_size=IF(file_size=0, VALUES(file_size), file_size),
file_type=IF(file_type='' OR file_type='file', VALUES(file_type), file_type), file_type=IF(file_type='' OR file_type='file', VALUES(file_type), file_type),
updated_at=updated_at`, updated_at=updated_at`,
file.GroupFileID, file.Topic, file.FileID, file.SourceTopic, file.SourceSeq, decodeUID(file.UploaderID), id, file.GroupFileID, file.Topic, file.FileID, file.SourceTopic, file.SourceSeq, decodeUID(file.UploaderID),
file.FileName, file.MIMEType, file.FileType, file.FileSize, file.Status, file.IdempotencyKey, now, now) file.FileName, file.MIMEType, file.FileType, file.FileSize, file.Status, file.IdempotencyKey, now, now)
if err != nil && !isDupe(err) { if err != nil && !isDupe(err) {
return nil, err return nil, err
@@ -1127,15 +1127,15 @@ func (a *adapter) TenantGroupFileCreate(id t.TenantID, file *t.GroupFile) (*t.Gr
} }
if fid := t.ParseUid(uploadID); !fid.IsZero() { if fid := t.ParseUid(uploadID); !fid.IsZero() {
if a.tenantScopedDB { if a.tenantScopedDB {
_, err = tx.ExecContext(ctx, `INSERT INTO im_filemsglinks(createdat,fileid,topic)
SELECT ?,?,? WHERE NOT EXISTS (
SELECT 1 FROM im_filemsglinks WHERE fileid=? AND topic=?)`,
now, decodeUID(fid), file.Topic, decodeUID(fid), file.Topic)
} else {
_, err = tx.ExecContext(ctx, `INSERT INTO im_filemsglinks(tenant_id,createdat,fileid,topic) _, err = tx.ExecContext(ctx, `INSERT INTO im_filemsglinks(tenant_id,createdat,fileid,topic)
SELECT ?,?,?,? WHERE NOT EXISTS ( SELECT ?,?,?,? WHERE NOT EXISTS (
SELECT 1 FROM im_filemsglinks WHERE tenant_id=? AND fileid=? AND topic=?)`, SELECT 1 FROM im_filemsglinks WHERE tenant_id=? AND fileid=? AND topic=?)`,
id, now, decodeUID(fid), file.Topic, id, decodeUID(fid), file.Topic) id, now, decodeUID(fid), file.Topic, id, decodeUID(fid), file.Topic)
} else {
_, err = tx.ExecContext(ctx, `INSERT INTO im_filemsglinks(createdat,fileid,topic)
SELECT ?,?,? WHERE NOT EXISTS (
SELECT 1 FROM im_filemsglinks WHERE fileid=? AND topic=?)`,
now, decodeUID(fid), file.Topic, decodeUID(fid), file.Topic)
} }
if err != nil { if err != nil {
return nil, err return nil, err
@@ -1144,7 +1144,7 @@ func (a *adapter) TenantGroupFileCreate(id t.TenantID, file *t.GroupFile) (*t.Gr
if err = tx.Commit(); err != nil { if err = tx.Commit(); err != nil {
return nil, err return nil, err
} }
created, err := a.groupFileByQuery(ctx, "SELECT * FROM im_group_files WHERE topic=? AND file_id=?", file.Topic, file.FileID) created, err := a.groupFileByQuery(ctx, "SELECT * FROM im_group_files WHERE tenant_id=? AND topic=? AND file_id=?", id, file.Topic, file.FileID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -1200,8 +1200,8 @@ func (a *adapter) TenantGroupFileDelete(id t.TenantID, topic, groupFileID string
} }
query := `UPDATE im_group_files query := `UPDATE im_group_files
SET status='deleted',deleted_by=?,deleted_at=?,updated_at=? SET status='deleted',deleted_by=?,deleted_at=?,updated_at=?
WHERE topic=? AND group_file_id=? AND status='active'` WHERE tenant_id=? AND topic=? AND group_file_id=? AND status='active'`
args := []any{decodeUID(actor), now, now, strings.TrimSpace(topic), strings.TrimSpace(groupFileID)} args := []any{decodeUID(actor), now, now, id, strings.TrimSpace(topic), strings.TrimSpace(groupFileID)}
if !canManage { if !canManage {
query += " AND uploader_id=?" query += " AND uploader_id=?"
args = append(args, decodeUID(actor)) args = append(args, decodeUID(actor))
@@ -1213,7 +1213,7 @@ func (a *adapter) TenantGroupFileDelete(id t.TenantID, topic, groupFileID string
if affected, _ := res.RowsAffected(); affected == 0 { if affected, _ := res.RowsAffected(); affected == 0 {
return nil, t.ErrNotFound return nil, t.ErrNotFound
} }
return a.groupFileByQuery(ctx, "SELECT * FROM im_group_files WHERE topic=? AND group_file_id=?", strings.TrimSpace(topic), strings.TrimSpace(groupFileID)) return a.groupFileByQuery(ctx, "SELECT * FROM im_group_files WHERE tenant_id=? AND topic=? AND group_file_id=?", id, strings.TrimSpace(topic), strings.TrimSpace(groupFileID))
} }
func (a *adapter) TenantGroupCreate(id t.TenantID, topic *t.Topic, owner t.Uid, ownerPrivate any, func (a *adapter) TenantGroupCreate(id t.TenantID, topic *t.Topic, owner t.Uid, ownerPrivate any,
@@ -1319,8 +1319,8 @@ func (a *adapter) TenantGroupInviteMember(id t.TenantID, topic string, inviter,
idempotencyKey += ":" + target.UserId() idempotencyKey += ":" + target.UserId()
} }
if _, err = tx.ExecContext(ctx, `INSERT INTO im_group_invitations if _, err = tx.ExecContext(ctx, `INSERT INTO im_group_invitations
(invitation_id,topic,inviter_user_id,target_user_id,message,status,result,idempotency_key,expires_at,handled_at,created_at,updated_at) (tenant_id,invitation_id,topic,inviter_user_id,target_user_id,message,status,result,idempotency_key,expires_at,handled_at,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`, invitationID, topic, decodeUID(inviter), decodeUID(target), message, VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, id, invitationID, topic, decodeUID(inviter), decodeUID(target), message,
t.GroupInvitationStatusJoined, "joined", idempotencyKey, now.Add(7*24*time.Hour), now, now, now); err != nil { t.GroupInvitationStatusJoined, "joined", idempotencyKey, now.Add(7*24*time.Hour), now, now, now); err != nil {
return false, err return false, err
} }
@@ -1490,8 +1490,8 @@ func (a *adapter) TenantGroupCanPost(id t.TenantID, topic string, uid t.Uid, adm
var muteAll, memberMuted bool var muteAll, memberMuted bool
err := a.db.QueryRowContext(ctx, `SELECT err := a.db.QueryRowContext(ctx, `SELECT
COALESCE((SELECT mute_all FROM im_group_settings WHERE tenant_id=? AND topic=? AND group_status=? LIMIT 1),0), COALESCE((SELECT mute_all FROM im_group_settings WHERE tenant_id=? AND topic=? AND group_status=? LIMIT 1),0),
EXISTS(SELECT 1 FROM im_group_mutes WHERE topic=? AND user_id=? AND status=? AND (expires_at IS NULL OR expires_at>?))`, EXISTS(SELECT 1 FROM im_group_mutes WHERE tenant_id=? AND topic=? AND user_id=? AND status=? AND (expires_at IS NULL OR expires_at>?))`,
id, topic, t.GroupStatusActive, topic, decodeUID(uid), t.GroupMuteStatusActive, now).Scan(&muteAll, &memberMuted) id, topic, t.GroupStatusActive, id, topic, decodeUID(uid), t.GroupMuteStatusActive, now).Scan(&muteAll, &memberMuted)
if err != nil { if err != nil {
return false, err return false, err
} }
@@ -1549,7 +1549,7 @@ func syncGroupMuteAllTx(tx *sqlx.Tx, tenantID t.TenantID, topic string, muteAll
mode = t.ModeCPublic &^ t.ModeWrite mode = t.ModeCPublic &^ t.ModeWrite
} }
_, err := tx.Exec(tenantShardSQL(tenantID, `UPDATE im_subscriptions AS s _, err := tx.Exec(tenantShardSQL(tenantID, `UPDATE im_subscriptions AS s
JOIN im_group_members AS gm ON gm.topic=s.topic AND gm.user_id=s.userid JOIN im_group_members AS gm ON gm.tenant_id=s.tenant_id AND gm.topic=s.topic AND gm.user_id=s.userid
SET s.modewant=?,s.modegiven=?,s.updatedat=?, SET s.modewant=?,s.modegiven=?,s.updatedat=?,
gm.version=gm.version+1,gm.updated_at=? gm.version=gm.version+1,gm.updated_at=?
WHERE s.topic=? AND s.deletedat IS NULL WHERE s.topic=? AND s.deletedat IS NULL
@@ -1611,7 +1611,7 @@ func (a *adapter) groupJoinRequestByQuery(ctx context.Context, query string, arg
func (a *adapter) groupInvitationByQuery(ctx context.Context, query string, args ...any) (*t.GroupInvitation, error) { func (a *adapter) groupInvitationByQuery(ctx context.Context, query string, args ...any) (*t.GroupInvitation, error) {
var item t.GroupInvitation var item t.GroupInvitation
var handledAt sql.NullTime var handledAt sql.NullTime
err := a.db.QueryRowxContext(ctx, query, args...).Scan(&item.ID, &item.InvitationID, &item.Topic, &item.InviterUserRaw, &item.TargetUserRaw, err := a.db.QueryRowxContext(ctx, query, args...).Scan(&item.TenantID, &item.ID, &item.InvitationID, &item.Topic, &item.InviterUserRaw, &item.TargetUserRaw,
&item.Message, &item.Status, &item.Result, &item.JoinRequestID, &item.IdempotencyKey, &item.ExpiresAt, &handledAt, &item.CreatedAt, &item.UpdatedAt) &item.Message, &item.Status, &item.Result, &item.JoinRequestID, &item.IdempotencyKey, &item.ExpiresAt, &handledAt, &item.CreatedAt, &item.UpdatedAt)
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
return nil, t.ErrNotFound return nil, t.ErrNotFound
server/store/types/groups.go
+5
View File
@@ -118,6 +118,7 @@ type GroupMemberListOptions struct {
} }
type GroupInvitation struct { type GroupInvitation struct {
TenantID TenantID `db:"tenant_id"`
ID int64 `db:"id"` ID int64 `db:"id"`
InvitationID string `db:"invitation_id"` InvitationID string `db:"invitation_id"`
Topic string `db:"topic"` Topic string `db:"topic"`
@@ -137,6 +138,7 @@ type GroupInvitation struct {
} }
type GroupJoinRequest struct { type GroupJoinRequest struct {
TenantID TenantID `db:"tenant_id"`
ID int64 `db:"id"` ID int64 `db:"id"`
RequestID string `db:"request_id"` RequestID string `db:"request_id"`
Topic string `db:"topic"` Topic string `db:"topic"`
@@ -160,6 +162,7 @@ type GroupJoinRequest struct {
} }
type GroupAnnouncement struct { type GroupAnnouncement struct {
TenantID TenantID `db:"tenant_id"`
ID int64 `db:"id"` ID int64 `db:"id"`
AnnouncementID string `db:"announcement_id"` AnnouncementID string `db:"announcement_id"`
Topic string `db:"topic"` Topic string `db:"topic"`
@@ -177,6 +180,7 @@ type GroupAnnouncement struct {
} }
type GroupMute struct { type GroupMute struct {
TenantID TenantID `db:"tenant_id"`
Topic string `db:"topic"` Topic string `db:"topic"`
UserID Uid `db:"-"` UserID Uid `db:"-"`
MutedBy Uid `db:"-"` MutedBy Uid `db:"-"`
@@ -195,6 +199,7 @@ type GroupMute struct {
} }
type GroupFile struct { type GroupFile struct {
TenantID TenantID `db:"tenant_id"`
ID int64 `db:"id"` ID int64 `db:"id"`
GroupFileID string `db:"group_file_id"` GroupFileID string `db:"group_file_id"`
Topic string `db:"topic"` Topic string `db:"topic"`