diff --git a/docs/IM-第一二章后续增强项.md b/docs/IM-第一二章后续增强项.md index b58999b..8005911 100644 --- a/docs/IM-第一二章后续增强项.md +++ b/docs/IM-第一二章后续增强项.md @@ -156,6 +156,7 @@ | 优先级 | 增强项 | 说明 | 建议验收点 | | --- | --- | --- | --- | | P0 | 群业务独立表拆分 | 当前群公告、进群申请、治理配置、禁言记录先存 Topic Aux，后续需要拆为独立表。 | 公告、申请、治理、禁言均有 tenant/topic 维度索引，可稳定分页和审计。 | +| P0 | 群业务表设计落地 | 已整理 `docs/IM-群聊业务表设计.sql`，后续需要按该设计接入迁移脚本和 store/adapter。 | 表结构审核通过后，群接口从 Topic Aux 切到业务表。 | | P0 | 群治理强事务状态机 | 邀请、申请审批、管理员变更、群主转让、禁言需在事务内同步业务表、订阅 ACL、成员数和事件。 | 并发审批/转让/禁言不会出现 ACL 与业务状态不一致。 | | P0 | 第六章幂等键落库 | 邀请、申请处理、公告发布、治理 PATCH、管理员批量、群主转让、禁言批量都需要统一幂等。 | 超时重试同一请求返回同一结果，不重复加人、升权或发事件。 | | P0 | 系统事件、Push 与 Outbox | 第六章所有成员/公告/治理变化需要生成可信群系统事件，并通过 Outbox 保证推送。 | 在线端实时刷新；模拟推送失败后可补偿且不重复。 | diff --git a/docs/IM-群聊业务表设计.sql b/docs/IM-群聊业务表设计.sql new file mode 100644 index 0000000..b49023b --- /dev/null +++ b/docs/IM-群聊业务表设计.sql @@ -0,0 +1,280 @@ +-- IM group business tables design. +-- Scope: chapter 6 group chat business truth tables. +-- Database: tenant database, default tinode. +-- Important: one tenant owns one database, so these tables intentionally do not contain tenant_id. +-- This SQL is for review first. Do not execute directly in production before confirming indexes and migration order. + +CREATE DATABASE IF NOT EXISTS tinode CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci; + +-- 1. Group profile and governance truth. +-- Topic/Subscription remains the communication projection; this table is the product business truth. +CREATE TABLE IF NOT EXISTS tinode.im_group_settings ( + 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, + member_upload_files TINYINT NOT NULL DEFAULT 1, + member_private_chat TINYINT NOT NULL DEFAULT 1, + member_invite TINYINT NOT NULL DEFAULT 1, + max_member_count INT NOT NULL DEFAULT 0 COMMENT '0 means use tenant default or unlimited', + max_admin_count INT NOT NULL DEFAULT 10, + member_count INT NOT NULL DEFAULT 0, + 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), + INDEX idx_im_group_settings_status_updated (group_status, updated_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 2. Group members and role truth. +-- Role changes must update this table and the Tinode Subscription ACL in the same transaction. +CREATE TABLE IF NOT EXISTS tinode.im_group_members ( + id BIGINT NOT NULL AUTO_INCREMENT, + topic VARCHAR(32) NOT NULL, + user_id BIGINT NOT NULL, + role VARCHAR(16) NOT NULL DEFAULT 'member', + member_status VARCHAR(16) NOT NULL DEFAULT 'active', + nickname VARCHAR(64) NOT NULL DEFAULT '', + pinned TINYINT NOT NULL DEFAULT 0, + mute_notifications TINYINT NOT NULL DEFAULT 0, + join_source VARCHAR(32) NOT NULL DEFAULT 'manual', + joined_by BIGINT, + joined_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + left_at DATETIME(3), + removed_by BIGINT, + 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_group_members_topic_user (topic, user_id), + INDEX idx_im_group_members_user_status (user_id, member_status, updated_at), + INDEX idx_im_group_members_topic_role (topic, role, member_status, id), + INDEX idx_im_group_members_topic_status (topic, member_status, id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 3. Direct invitations. +-- If join_approval is off, accepting an invitation can directly create im_group_members + Subscription. +-- If join_approval is on, accepting may create a pending im_group_join_requests record. +CREATE TABLE IF NOT EXISTS tinode.im_group_invitations ( + id BIGINT NOT NULL AUTO_INCREMENT, + invitation_id VARCHAR(64) NOT NULL, + topic VARCHAR(32) NOT NULL, + inviter_user_id BIGINT NOT NULL, + target_user_id BIGINT NOT NULL, + message VARCHAR(256) NOT NULL DEFAULT '', + status VARCHAR(16) NOT NULL DEFAULT 'pending', + result VARCHAR(32) NOT NULL DEFAULT '', + join_request_id VARCHAR(64) NOT NULL DEFAULT '', + idempotency_key VARCHAR(128) NOT NULL DEFAULT '', + expires_at DATETIME(3) NOT NULL, + handled_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 (id), + UNIQUE INDEX uk_im_group_invitations_invitation (invitation_id), + UNIQUE INDEX uk_im_group_invitations_idempotency (inviter_user_id, idempotency_key), + INDEX idx_im_group_invitations_target (target_user_id, status, expires_at), + INDEX idx_im_group_invitations_topic_status (topic, status, id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 4. Join requests. +-- Pending users must not have a readable Subscription before approval. +CREATE TABLE IF NOT EXISTS tinode.im_group_join_requests ( + id BIGINT NOT NULL AUTO_INCREMENT, + request_id VARCHAR(64) NOT NULL, + topic VARCHAR(32) NOT NULL, + applicant_user_id BIGINT NOT NULL, + inviter_user_id BIGINT, + source VARCHAR(32) NOT NULL DEFAULT 'manual', + qr_token_hash CHAR(64) NOT NULL DEFAULT '', + message VARCHAR(256) NOT NULL DEFAULT '', + status VARCHAR(16) NOT NULL DEFAULT 'pending', + reject_reason VARCHAR(256) NOT NULL DEFAULT '', + handled_by BIGINT, + handled_at DATETIME(3), + expires_at DATETIME(3), + idempotency_key VARCHAR(128) 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_group_join_requests_request (request_id), + UNIQUE INDEX uk_im_group_join_requests_idempotency (applicant_user_id, idempotency_key), + INDEX idx_im_group_join_requests_topic_status (topic, status, id), + INDEX idx_im_group_join_requests_applicant (applicant_user_id, topic, status, id), + INDEX idx_im_group_join_requests_expire (status, expires_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 5. Group announcements. +CREATE TABLE IF NOT EXISTS tinode.im_group_announcements ( + id BIGINT NOT NULL AUTO_INCREMENT, + announcement_id VARCHAR(64) NOT NULL, + topic VARCHAR(32) NOT NULL, + author_user_id BIGINT NOT NULL, + content TEXT NOT NULL, + status VARCHAR(16) NOT NULL DEFAULT 'current', + version INT NOT NULL DEFAULT 1, + published_at DATETIME(3) NOT NULL, + deleted_by BIGINT, + deleted_at DATETIME(3), + idempotency_key VARCHAR(128) 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 (id), + UNIQUE INDEX uk_im_group_announcements_announcement (announcement_id), + UNIQUE INDEX uk_im_group_announcements_idempotency (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_time (topic, published_at DESC, id DESC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 6. Announcement read receipts. +CREATE TABLE IF NOT EXISTS tinode.im_group_announcement_reads ( + 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) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 7. Current member mute state. +-- Historical changes are stored in im_group_admin_logs. +CREATE TABLE IF NOT EXISTS tinode.im_group_mutes ( + 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), + INDEX idx_im_group_mutes_user_status (user_id, status, expires_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 8. Group QR codes / invite tokens. +-- qr_token should remain opaque to clients. Store only token hash. +CREATE TABLE IF NOT EXISTS tinode.im_group_qrcodes ( + id BIGINT NOT NULL AUTO_INCREMENT, + token_id VARCHAR(64) NOT NULL, + token_hash CHAR(64) NOT NULL, + topic VARCHAR(32) NOT NULL, + created_by BIGINT NOT NULL, + status VARCHAR(16) NOT NULL DEFAULT 'active', + expires_at DATETIME(3) NOT NULL, + revoked_by BIGINT, + revoked_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 (id), + UNIQUE INDEX uk_im_group_qrcodes_token (token_id), + UNIQUE INDEX uk_im_group_qrcodes_hash (token_hash), + INDEX idx_im_group_qrcodes_topic_status (topic, status, expires_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 9. Group file directory. +-- File binary metadata still lives in the existing file table/storage. This is the group-visible index. +CREATE TABLE IF NOT EXISTS tinode.im_group_files ( + id BIGINT NOT NULL AUTO_INCREMENT, + group_file_id VARCHAR(64) NOT NULL, + topic VARCHAR(32) NOT NULL, + file_id VARCHAR(64) NOT NULL, + source_topic VARCHAR(32) NOT NULL DEFAULT '', + source_seq INT NOT NULL DEFAULT 0, + uploader_id BIGINT NOT NULL, + file_name VARCHAR(255) NOT NULL DEFAULT '', + mime_type VARCHAR(128) NOT NULL DEFAULT '', + file_type VARCHAR(32) NOT NULL DEFAULT 'file', + file_size BIGINT NOT NULL DEFAULT 0, + status VARCHAR(16) NOT NULL DEFAULT 'active', + deleted_by BIGINT, + deleted_at DATETIME(3), + idempotency_key VARCHAR(128) 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 (id), + UNIQUE INDEX uk_im_group_files_group_file (group_file_id), + UNIQUE INDEX uk_im_group_files_topic_file (topic, file_id), + UNIQUE INDEX uk_im_group_files_idempotency (topic, uploader_id, idempotency_key), + INDEX idx_im_group_files_topic_type (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) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 10. Group operation audit logs. +-- Store role changes, governance changes, mutes, announcements, owner transfer, dissolve, file delete, etc. +CREATE TABLE IF NOT EXISTS tinode.im_group_admin_logs ( + id BIGINT NOT NULL AUTO_INCREMENT, + log_id VARCHAR(64) NOT NULL, + topic VARCHAR(32) NOT NULL, + operator_id BIGINT NOT NULL, + target_user_id BIGINT, + action VARCHAR(64) NOT NULL, + 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 '', + 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), + INDEX idx_im_group_admin_logs_topic_time (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) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 11. Group operation idempotency. +-- This can later be replaced by a global HTTP idempotency table. +CREATE TABLE IF NOT EXISTS tinode.im_group_idempotency ( + id BIGINT NOT NULL AUTO_INCREMENT, + user_id BIGINT NOT NULL, + method VARCHAR(8) NOT NULL, + route VARCHAR(256) NOT NULL, + idempotency_key VARCHAR(128) NOT NULL, + body_hash CHAR(64) NOT NULL, + result_code INT NOT NULL DEFAULT 0, + result_body JSON, + 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_group_idempotency_request (user_id, method, route, idempotency_key), + INDEX idx_im_group_idempotency_created (created_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 12. Group outbox events. +-- Transactional bridge for group.membership.updated, group.join_request.created, +-- group.announcement.updated, group.role.updated, group.mute.updated, group.dissolved, etc. +CREATE TABLE IF NOT EXISTS tinode.im_group_events_outbox ( + id BIGINT NOT NULL AUTO_INCREMENT, + event_id VARCHAR(64) NOT NULL, + topic VARCHAR(32) NOT NULL, + actor_user_id BIGINT NOT NULL DEFAULT 0, + event_type VARCHAR(64) NOT NULL, + payload JSON, + status VARCHAR(16) NOT NULL DEFAULT 'pending', + attempts INT NOT NULL DEFAULT 0, + next_retry_at DATETIME(3), + delivered_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 (id), + UNIQUE INDEX uk_im_group_events_outbox_event (event_id), + INDEX idx_im_group_events_outbox_pending (status, next_retry_at, id), + INDEX idx_im_group_events_outbox_topic (topic, id DESC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; diff --git a/scripts/apply_group_business_tables.go b/scripts/apply_group_business_tables.go new file mode 100644 index 0000000..f38dea9 --- /dev/null +++ b/scripts/apply_group_business_tables.go @@ -0,0 +1,59 @@ +//go:build ignore + +package main + +import ( + "database/sql" + "fmt" + "os" + "strings" + + _ "github.com/go-sql-driver/mysql" +) + +func main() { + dsn := strings.TrimPrefix(os.Getenv("IM_MYSQL_DSN"), "mysql://") + if dsn == "" { + fatal("IM_MYSQL_DSN is required") + } + db, err := sql.Open("mysql", dsn) + if err != nil { + fatal(err) + } + defer db.Close() + if err = db.Ping(); err != nil { + fatal(err) + } + + raw, err := os.ReadFile("docs/IM-群聊业务表设计.sql") + if err != nil { + fatal(err) + } + sqlText := stripSQLLineComments(string(raw)) + for _, stmt := range strings.Split(sqlText, ";") { + stmt = strings.TrimSpace(stmt) + if stmt == "" { + continue + } + if _, err = db.Exec(stmt); err != nil { + fatal(fmt.Errorf("exec failed: %s: %w", stmt, err)) + } + } + fmt.Println("group business tables applied") +} + +func stripSQLLineComments(text string) string { + var lines []string + for _, line := range strings.Split(text, "\n") { + if strings.HasPrefix(strings.TrimSpace(line), "--") { + continue + } + lines = append(lines, line) + } + return strings.Join(lines, "\n") +} + +func fatal(v any) { + fmt.Fprintln(os.Stderr, v) + os.Exit(1) +} diff --git a/scripts/check_group_business_tables.go b/scripts/check_group_business_tables.go new file mode 100644 index 0000000..9654dbe --- /dev/null +++ b/scripts/check_group_business_tables.go @@ -0,0 +1,59 @@ +//go:build ignore + +package main + +import ( + "database/sql" + "fmt" + "os" + "strings" + + _ "github.com/go-sql-driver/mysql" +) + +func main() { + dsn := strings.TrimPrefix(os.Getenv("IM_MYSQL_DSN"), "mysql://") + if dsn == "" { + fatal("IM_MYSQL_DSN is required") + } + db, err := sql.Open("mysql", dsn) + if err != nil { + fatal(err) + } + defer db.Close() + if err = db.Ping(); err != nil { + fatal(err) + } + + tables := []string{ + "im_group_settings", + "im_group_members", + "im_group_invitations", + "im_group_join_requests", + "im_group_announcements", + "im_group_announcement_reads", + "im_group_mutes", + "im_group_qrcodes", + "im_group_files", + "im_group_admin_logs", + "im_group_idempotency", + "im_group_events_outbox", + } + for _, table := range tables { + var count int + err = db.QueryRow(`SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='tinode' AND TABLE_NAME=?`, table).Scan(&count) + if err != nil { + fatal(err) + } + status := "missing" + if count == 1 { + status = "ok" + } + fmt.Printf("%s: %s\n", table, status) + } +} + +func fatal(v any) { + fmt.Fprintln(os.Stderr, v) + os.Exit(1) +} diff --git a/server/db/adapter.go b/server/db/adapter.go index 3013775..de7632b 100644 --- a/server/db/adapter.go +++ b/server/db/adapter.go @@ -294,6 +294,24 @@ type TenantBusinessAdapter interface { TenantCallFinish(t.TenantID, string, int, string, int64, time.Time) error TenantCallList(t.TenantID, t.Uid, string, string, int) ([]t.CallRecord, string, error) + TenantGroupSettingsCreate(t.TenantID, *t.GroupSettings) (*t.GroupSettings, error) + TenantGroupSettingsGet(t.TenantID, string) (*t.GroupSettings, error) + TenantGroupSettingsUpdate(t.TenantID, string, t.GroupSettingsPatch) (*t.GroupSettings, error) + TenantGroupMemberUpsert(t.TenantID, *t.GroupMember) (*t.GroupMember, error) + TenantGroupMemberGet(t.TenantID, string, t.Uid) (*t.GroupMember, error) + TenantGroupMemberList(t.TenantID, string, t.GroupMemberListOptions) ([]t.GroupMember, string, error) + TenantGroupMemberUpdate(t.TenantID, string, t.Uid, t.GroupMemberPatch) (*t.GroupMember, error) + TenantGroupInvitationCreate(t.TenantID, *t.GroupInvitation) (*t.GroupInvitation, error) + TenantGroupJoinRequestCreate(t.TenantID, *t.GroupJoinRequest) (*t.GroupJoinRequest, error) + TenantGroupJoinRequestList(t.TenantID, string, string, string, int) ([]t.GroupJoinRequest, string, error) + TenantGroupJoinRequestResolve(t.TenantID, string, string, t.Uid, string, string, int, time.Time) (*t.GroupJoinRequest, error) + TenantGroupAnnouncementLatest(t.TenantID, string) (*t.GroupAnnouncement, error) + TenantGroupAnnouncementPut(t.TenantID, *t.GroupAnnouncement, int) (*t.GroupAnnouncement, error) + TenantGroupAnnouncementRead(t.TenantID, string, string, t.Uid, time.Time) error + TenantGroupMuteList(t.TenantID, string, string, int) ([]t.GroupMute, string, error) + TenantGroupMuteUpsert(t.TenantID, *t.GroupMute) (*t.GroupMute, error) + TenantGroupMuteDelete(t.TenantID, string, t.Uid, t.Uid, time.Time) error + TenantWorkbenchModules(t.TenantID, t.Uid) ([]t.WorkbenchModule, int, error) TenantOrganizationDepartments(t.TenantID, t.Uid, string, string, int) ([]t.OrganizationDepartment, string, error) TenantOrganizationMembers(t.TenantID, t.Uid, string, string, int) ([]t.OrganizationMember, string, error) diff --git a/server/db/mysql/tenant_groups.go b/server/db/mysql/tenant_groups.go new file mode 100644 index 0000000..dbe4b4a --- /dev/null +++ b/server/db/mysql/tenant_groups.go @@ -0,0 +1,760 @@ +//go:build mysql +// +build mysql + +package mysql + +import ( + "context" + "database/sql" + "strconv" + "strings" + "time" + + "github.com/tinode/chat/server/store" + t "github.com/tinode/chat/server/store/types" +) + +func (a *adapter) TenantGroupSettingsCreate(id t.TenantID, settings *t.GroupSettings) (*t.GroupSettings, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupSettingsCreate(id, settings) + } + if settings == nil || settings.Topic == "" || settings.OwnerUserID.IsZero() { + return nil, t.ErrMalformed + } + now := time.Now().UTC().Round(time.Millisecond) + if settings.CreatedAt.IsZero() { + settings.CreatedAt = now + } + if settings.UpdatedAt.IsZero() { + settings.UpdatedAt = settings.CreatedAt + } + if settings.GroupStatus == "" { + settings.GroupStatus = t.GroupStatusActive + } + if settings.MaxAdminCount <= 0 { + settings.MaxAdminCount = 10 + } + if settings.Version <= 0 { + settings.Version = 1 + } + ctx, cancel := a.getContext() + 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(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + 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, + 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 { + return nil, err + } + return a.TenantGroupSettingsGet(id, settings.Topic) +} + +func (a *adapter) TenantGroupSettingsGet(id t.TenantID, topic string) (*t.GroupSettings, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupSettingsGet(id, topic) + } + if topic == "" { + return nil, t.ErrMalformed + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + return a.groupSettingsByTopic(ctx, topic) +} + +func (a *adapter) TenantGroupSettingsUpdate(id t.TenantID, topic string, patch t.GroupSettingsPatch) (*t.GroupSettings, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupSettingsUpdate(id, topic, patch) + } + if topic == "" { + return nil, t.ErrMalformed + } + set := []string{"updated_at=?"} + args := []any{time.Now().UTC().Round(time.Millisecond)} + if patch.Name != nil { + set = append(set, "name=?") + args = append(args, *patch.Name) + } + if patch.AvatarFileID != nil { + set = append(set, "avatar_file_id=?") + args = append(args, *patch.AvatarFileID) + } + if patch.Description != nil { + set = append(set, "description=?") + args = append(args, *patch.Description) + } + if patch.OwnerUserID != nil { + set = append(set, "owner_user_id=?") + args = append(args, decodeUID(*patch.OwnerUserID)) + } + if patch.JoinApproval != nil { + set = append(set, "join_approval=?") + args = append(args, *patch.JoinApproval) + } + if patch.MuteAll != nil { + set = append(set, "mute_all=?") + args = append(args, *patch.MuteAll) + } + if patch.MemberUploadFiles != nil { + set = append(set, "member_upload_files=?") + args = append(args, *patch.MemberUploadFiles) + } + if patch.MemberPrivateChat != nil { + set = append(set, "member_private_chat=?") + args = append(args, *patch.MemberPrivateChat) + } + if patch.MemberInvite != nil { + set = append(set, "member_invite=?") + args = append(args, *patch.MemberInvite) + } + if patch.MemberCountDelta != 0 { + set = append(set, "member_count=GREATEST(member_count+?,0)") + args = append(args, patch.MemberCountDelta) + } + set = append(set, "version=version+1") + args = append(args, topic) + where := "topic=?" + if patch.Version > 0 { + where += " AND version=?" + args = append(args, patch.Version) + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + res, err := a.db.ExecContext(ctx, "UPDATE im_group_settings SET "+strings.Join(set, ",")+" WHERE "+where, args...) + if err != nil { + return nil, err + } + if affected, _ := res.RowsAffected(); affected == 0 { + if patch.Version > 0 { + return nil, t.ErrInvalidResponse + } + return nil, t.ErrNotFound + } + return a.groupSettingsByTopic(ctx, topic) +} + +func (a *adapter) TenantGroupMemberUpsert(id t.TenantID, member *t.GroupMember) (*t.GroupMember, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupMemberUpsert(id, member) + } + if member == nil || member.Topic == "" || member.UserID.IsZero() { + return nil, t.ErrMalformed + } + now := time.Now().UTC().Round(time.Millisecond) + if member.JoinedAt.IsZero() { + member.JoinedAt = now + } + if member.CreatedAt.IsZero() { + member.CreatedAt = now + } + if member.UpdatedAt.IsZero() { + member.UpdatedAt = now + } + if member.Role == "" { + member.Role = t.GroupMemberRoleMember + } + if member.MemberStatus == "" { + member.MemberStatus = t.GroupMemberStatusActive + } + if member.JoinSource == "" { + member.JoinSource = "manual" + } + if member.Version <= 0 { + member.Version = 1 + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + _, err := a.db.ExecContext(ctx, `INSERT INTO im_group_members + (topic,user_id,role,member_status,nickname,pinned,mute_notifications,join_source,joined_by,joined_at,version,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?) + ON DUPLICATE KEY UPDATE role=VALUES(role),member_status=VALUES(member_status),nickname=VALUES(nickname),pinned=VALUES(pinned),mute_notifications=VALUES(mute_notifications),join_source=VALUES(join_source),joined_by=VALUES(joined_by),joined_at=IF(member_status='active',joined_at,VALUES(joined_at)),left_at=NULL,removed_by=NULL,version=version+1,updated_at=VALUES(updated_at)`, + member.Topic, decodeUID(member.UserID), member.Role, member.MemberStatus, member.Nickname, member.Pinned, member.MuteNotifications, + member.JoinSource, nullableUID(member.JoinedBy), member.JoinedAt, member.Version, member.CreatedAt, member.UpdatedAt) + if err != nil { + return nil, err + } + _ = a.recountGroupMembers(ctx, member.Topic) + return a.TenantGroupMemberGet(id, member.Topic, member.UserID) +} + +func (a *adapter) TenantGroupMemberGet(id t.TenantID, topic string, uid t.Uid) (*t.GroupMember, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupMemberGet(id, topic, uid) + } + if topic == "" || uid.IsZero() { + return nil, t.ErrMalformed + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + return a.groupMemberByQuery(ctx, "SELECT * FROM im_group_members WHERE topic=? AND user_id=?", topic, decodeUID(uid)) +} + +func (a *adapter) TenantGroupMemberList(id t.TenantID, topic string, opts t.GroupMemberListOptions) ([]t.GroupMember, string, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, "", err + } else if tenantAdp != a { + return tenantAdp.TenantGroupMemberList(id, topic, opts) + } + if topic == "" { + return nil, "", t.ErrMalformed + } + if opts.Limit <= 0 || opts.Limit > 100 { + opts.Limit = 30 + } + where := []string{"topic=?", "member_status=?"} + args := []any{topic, t.GroupMemberStatusActive} + if opts.Role != "" { + where = append(where, "role=?") + args = append(args, opts.Role) + } + if opts.Query != "" { + where = append(where, "nickname LIKE ?") + args = append(args, "%"+opts.Query+"%") + } + if opts.Cursor != "" { + parsed, err := strconv.ParseInt(opts.Cursor, 10, 64) + if err != nil || parsed <= 0 { + return nil, "", t.ErrMalformed + } + where = append(where, "id>?") + args = append(args, parsed) + } + args = append(args, opts.Limit+1) + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + rows, err := a.db.QueryxContext(ctx, "SELECT * FROM im_group_members WHERE "+strings.Join(where, " AND ")+" ORDER BY id LIMIT ?", args...) + if err != nil { + return nil, "", err + } + defer rows.Close() + items := make([]t.GroupMember, 0, opts.Limit) + next := "" + for rows.Next() { + var item t.GroupMember + if err = rows.StructScan(&item); err != nil { + return nil, "", err + } + scanGroupMember(&item) + if len(items) == opts.Limit { + next = strconv.FormatInt(item.ID, 10) + continue + } + items = append(items, item) + } + return items, next, rows.Err() +} + +func (a *adapter) TenantGroupMemberUpdate(id t.TenantID, topic string, uid t.Uid, patch t.GroupMemberPatch) (*t.GroupMember, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupMemberUpdate(id, topic, uid, patch) + } + set := []string{"updated_at=?", "version=version+1"} + args := []any{time.Now().UTC().Round(time.Millisecond)} + if patch.Role != nil { + set = append(set, "role=?") + args = append(args, *patch.Role) + } + if patch.MemberStatus != nil { + set = append(set, "member_status=?") + args = append(args, *patch.MemberStatus) + } + if patch.Nickname != nil { + set = append(set, "nickname=?") + args = append(args, *patch.Nickname) + } + if patch.Pinned != nil { + set = append(set, "pinned=?") + args = append(args, *patch.Pinned) + } + if patch.MuteNotifications != nil { + set = append(set, "mute_notifications=?") + args = append(args, *patch.MuteNotifications) + } + if patch.LeftAt != nil { + set = append(set, "left_at=?") + args = append(args, *patch.LeftAt) + } + if patch.RemovedBy != nil { + set = append(set, "removed_by=?") + args = append(args, nullableUID(*patch.RemovedBy)) + } + args = append(args, topic, decodeUID(uid)) + where := "topic=? AND user_id=?" + if patch.Version > 0 { + where += " AND version=?" + args = append(args, patch.Version) + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + res, err := a.db.ExecContext(ctx, "UPDATE im_group_members SET "+strings.Join(set, ",")+" WHERE "+where, args...) + if err != nil { + return nil, err + } + if affected, _ := res.RowsAffected(); affected == 0 { + if patch.Version > 0 { + return nil, t.ErrInvalidResponse + } + return nil, t.ErrNotFound + } + _ = a.recountGroupMembers(ctx, topic) + return a.groupMemberByQuery(ctx, "SELECT * FROM im_group_members WHERE topic=? AND user_id=?", topic, decodeUID(uid)) +} + +func (a *adapter) TenantGroupInvitationCreate(id t.TenantID, inv *t.GroupInvitation) (*t.GroupInvitation, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupInvitationCreate(id, inv) + } + now := time.Now().UTC().Round(time.Millisecond) + if inv.InvitationID == "" { + inv.InvitationID = "ginv_" + store.Store.GetUidString() + } + if inv.IdempotencyKey == "" { + inv.IdempotencyKey = inv.InvitationID + } + if inv.Status == "" { + inv.Status = t.GroupInvitationStatusJoined + } + if inv.ExpiresAt.IsZero() { + inv.ExpiresAt = now.Add(7 * 24 * time.Hour) + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + _, err := a.db.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) + 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) + if err != nil { + return nil, err + } + return inv, nil +} + +func (a *adapter) TenantGroupJoinRequestCreate(id t.TenantID, req *t.GroupJoinRequest) (*t.GroupJoinRequest, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupJoinRequestCreate(id, req) + } + now := time.Now().UTC().Round(time.Millisecond) + if req.Status == "" { + req.Status = t.GroupJoinRequestStatusPending + } + if req.Source == "" { + req.Source = "manual" + } + if req.Version <= 0 { + req.Version = 1 + } + if req.IdempotencyKey == "" { + req.IdempotencyKey = req.RequestID + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + _, err := a.db.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) + 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) + if err != nil { + return nil, err + } + return a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE request_id=?", req.RequestID) +} + +func (a *adapter) TenantGroupJoinRequestList(id t.TenantID, topic, status, cursor string, limit int) ([]t.GroupJoinRequest, string, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, "", err + } else if tenantAdp != a { + return tenantAdp.TenantGroupJoinRequestList(id, topic, status, cursor, limit) + } + where := []string{"topic=?"} + args := []any{topic} + if status != "" && status != "all" { + where = append(where, "status=?") + args = append(args, status) + } + if cursor != "" { + parsed, err := strconv.ParseInt(cursor, 10, 64) + if err != nil || parsed <= 0 { + return nil, "", t.ErrMalformed + } + where = append(where, "id>?") + args = append(args, parsed) + } + args = append(args, limit+1) + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + rows, err := a.db.QueryxContext(ctx, "SELECT * FROM im_group_join_requests WHERE "+strings.Join(where, " AND ")+" ORDER BY id LIMIT ?", args...) + if err != nil { + return nil, "", err + } + defer rows.Close() + items := make([]t.GroupJoinRequest, 0, limit) + next := "" + for rows.Next() { + var item t.GroupJoinRequest + if err = rows.StructScan(&item); err != nil { + return nil, "", err + } + scanGroupJoinRequest(&item) + if len(items) == limit { + next = strconv.FormatInt(item.ID, 10) + continue + } + items = append(items, item) + } + return items, next, rows.Err() +} + +func (a *adapter) TenantGroupJoinRequestResolve(id t.TenantID, topic, requestID string, actor t.Uid, action, rejectReason string, version int, now time.Time) (*t.GroupJoinRequest, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupJoinRequestResolve(id, topic, requestID, actor, action, rejectReason, version, now) + } + status := t.GroupJoinRequestStatusRejected + if action == "accept" { + status = t.GroupJoinRequestStatusAccepted + } + args := []any{status, rejectReason, decodeUID(actor), now, now, topic, requestID, t.GroupJoinRequestStatusPending} + where := "topic=? AND request_id=? AND status=?" + if version > 0 { + where += " AND version=?" + args = append(args, version) + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + res, err := a.db.ExecContext(ctx, "UPDATE im_group_join_requests SET status=?,reject_reason=?,handled_by=?,handled_at=?,updated_at=?,version=version+1 WHERE "+where, args...) + if err != nil { + return nil, err + } + if affected, _ := res.RowsAffected(); affected == 0 { + if version > 0 { + return nil, t.ErrInvalidResponse + } + return nil, t.ErrNotFound + } + return a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE topic=? AND request_id=?", topic, requestID) +} + +func (a *adapter) TenantGroupAnnouncementLatest(id t.TenantID, topic string) (*t.GroupAnnouncement, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupAnnouncementLatest(id, topic) + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + return a.groupAnnouncementByQuery(ctx, "SELECT * FROM im_group_announcements WHERE topic=? AND status=? ORDER BY version DESC,id DESC LIMIT 1", topic, t.GroupAnnouncementStatusCurrent) +} + +func (a *adapter) TenantGroupAnnouncementPut(id t.TenantID, ann *t.GroupAnnouncement, expectedVersion int) (*t.GroupAnnouncement, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupAnnouncementPut(id, ann, expectedVersion) + } + 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() + } + }() + var currentVersion int + err = tx.QueryRowContext(ctx, "SELECT COALESCE(MAX(version),0) FROM im_group_announcements WHERE topic=?", ann.Topic).Scan(¤tVersion) + if err != nil { + return nil, err + } + if expectedVersion > 0 && currentVersion > 0 && expectedVersion != currentVersion { + return nil, t.ErrInvalidResponse + } + version := currentVersion + 1 + now := ann.PublishedAt + if now.IsZero() { + 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 { + return nil, err + } + if ann.AnnouncementID == "" { + ann.AnnouncementID = "ann_" + store.Store.GetUidString() + } + if ann.IdempotencyKey == "" { + 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) + VALUES(?,?,?,?,?,?,?,?,?,?)`, ann.AnnouncementID, ann.Topic, decodeUID(ann.AuthorUserID), ann.Content, t.GroupAnnouncementStatusCurrent, version, now, ann.IdempotencyKey, now, now); err != nil { + return nil, err + } + if err = tx.Commit(); err != nil { + return nil, err + } + return a.TenantGroupAnnouncementLatest(id, ann.Topic) +} + +func (a *adapter) TenantGroupAnnouncementRead(id t.TenantID, topic, announcementID string, uid t.Uid, readAt time.Time) error { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return err + } else if tenantAdp != a { + return tenantAdp.TenantGroupAnnouncementRead(id, topic, announcementID, uid, readAt) + } + ctx, cancel := a.getContext() + 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) + return err +} + +func (a *adapter) TenantGroupMuteList(id t.TenantID, topic, cursor string, limit int) ([]t.GroupMute, string, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, "", err + } else if tenantAdp != a { + return tenantAdp.TenantGroupMuteList(id, topic, cursor, limit) + } + where := []string{"topic=?", "status=?"} + args := []any{topic, t.GroupMuteStatusActive} + if cursor != "" { + parsed, err := strconv.ParseInt(cursor, 10, 64) + if err != nil || parsed <= 0 { + return nil, "", t.ErrMalformed + } + where = append(where, "user_id>?") + args = append(args, parsed) + } + args = append(args, limit+1) + ctx, cancel := a.getContext() + 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...) + if err != nil { + return nil, "", err + } + defer rows.Close() + items := make([]t.GroupMute, 0, limit) + next := "" + for rows.Next() { + var item t.GroupMute + if err = rows.StructScan(&item); err != nil { + return nil, "", err + } + scanGroupMute(&item) + if len(items) == limit { + next = strconv.FormatInt(item.UserIDRaw, 10) + continue + } + items = append(items, item) + } + return items, next, rows.Err() +} + +func (a *adapter) TenantGroupMuteUpsert(id t.TenantID, mute *t.GroupMute) (*t.GroupMute, error) { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return nil, err + } else if tenantAdp != a { + return tenantAdp.TenantGroupMuteUpsert(id, mute) + } + now := mute.MutedAt + if now.IsZero() { + now = time.Now().UTC().Round(time.Millisecond) + } + if mute.Status == "" { + mute.Status = t.GroupMuteStatusActive + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + _, err := a.db.ExecContext(ctx, `INSERT INTO im_group_mutes(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) + if err != nil { + return nil, err + } + return a.groupMuteByQuery(ctx, "SELECT * 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 { + if tenantAdp, err := a.routeGroupTenant(id); err != nil { + return err + } else if tenantAdp != a { + return tenantAdp.TenantGroupMuteDelete(id, topic, uid, actor, now) + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + res, err := a.db.ExecContext(ctx, "UPDATE im_group_mutes SET status=?,revoked_by=?,revoked_at=?,version=version+1,updated_at=? WHERE topic=? AND user_id=? AND status=?", + t.GroupMuteStatusRevoked, decodeUID(actor), now, now, topic, decodeUID(uid), t.GroupMuteStatusActive) + if err != nil { + return err + } + if affected, _ := res.RowsAffected(); affected == 0 { + return t.ErrNotFound + } + return nil +} + +func (a *adapter) routeGroupTenant(id t.TenantID) (*adapter, error) { + if !a.tenantScopedDB { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil { + return nil, err + } + return tenantAdp, nil + } + return a, validTenant(id) +} + +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) + if err == sql.ErrNoRows { + return nil, t.ErrNotFound + } + if err != nil { + return nil, err + } + scanGroupSettings(&item) + return &item, nil +} + +func (a *adapter) groupMemberByQuery(ctx context.Context, query string, args ...any) (*t.GroupMember, error) { + var item t.GroupMember + err := a.db.QueryRowxContext(ctx, query, args...).StructScan(&item) + if err == sql.ErrNoRows { + return nil, t.ErrNotFound + } + if err != nil { + return nil, err + } + scanGroupMember(&item) + return &item, nil +} + +func (a *adapter) groupJoinRequestByQuery(ctx context.Context, query string, args ...any) (*t.GroupJoinRequest, error) { + var item t.GroupJoinRequest + err := a.db.QueryRowxContext(ctx, query, args...).StructScan(&item) + if err == sql.ErrNoRows { + return nil, t.ErrNotFound + } + if err != nil { + return nil, err + } + scanGroupJoinRequest(&item) + return &item, nil +} + +func (a *adapter) groupAnnouncementByQuery(ctx context.Context, query string, args ...any) (*t.GroupAnnouncement, error) { + var item t.GroupAnnouncement + err := a.db.QueryRowxContext(ctx, query, args...).StructScan(&item) + if err == sql.ErrNoRows { + return nil, t.ErrNotFound + } + if err != nil { + return nil, err + } + item.AuthorUserID = store.EncodeUid(item.AuthorUserIDRaw) + return &item, nil +} + +func (a *adapter) groupMuteByQuery(ctx context.Context, query string, args ...any) (*t.GroupMute, error) { + var item t.GroupMute + err := a.db.QueryRowxContext(ctx, query, args...).StructScan(&item) + if err == sql.ErrNoRows { + return nil, t.ErrNotFound + } + if err != nil { + return nil, err + } + scanGroupMute(&item) + return &item, nil +} + +func (a *adapter) recountGroupMembers(ctx context.Context, topic string) error { + _, err := a.db.ExecContext(ctx, "UPDATE im_group_settings SET member_count=(SELECT COUNT(*) FROM im_group_members WHERE topic=? AND member_status=?),updated_at=CURRENT_TIMESTAMP(3) WHERE topic=?", topic, t.GroupMemberStatusActive, topic) + return err +} + +func scanGroupSettings(item *t.GroupSettings) { + item.OwnerUserID = store.EncodeUid(item.OwnerUserIDRaw) + item.CreatedBy = store.EncodeUid(item.CreatedByRaw) + if item.DissolvedByRaw != nil { + item.DissolvedBy = store.EncodeUid(*item.DissolvedByRaw) + } +} + +func scanGroupMember(item *t.GroupMember) { + item.UserID = store.EncodeUid(item.UserIDRaw) + if item.JoinedByRaw != nil { + item.JoinedBy = store.EncodeUid(*item.JoinedByRaw) + } + if item.RemovedByRaw != nil { + item.RemovedBy = store.EncodeUid(*item.RemovedByRaw) + } +} + +func scanGroupJoinRequest(item *t.GroupJoinRequest) { + item.ApplicantUserID = store.EncodeUid(item.ApplicantUserRaw) + if item.InviterUserRaw != nil { + item.InviterUserID = store.EncodeUid(*item.InviterUserRaw) + } + if item.HandledByRaw != nil { + item.HandledBy = store.EncodeUid(*item.HandledByRaw) + } +} + +func scanGroupMute(item *t.GroupMute) { + item.UserID = store.EncodeUid(item.UserIDRaw) + item.MutedBy = store.EncodeUid(item.MutedByRaw) + if item.RevokedRaw != nil { + item.RevokedBy = store.EncodeUid(*item.RevokedRaw) + } +} diff --git a/server/im_http.go b/server/im_http.go index 8da592a..cac2c3d 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -2956,6 +2956,40 @@ func serveIMGroupCreate(wrt http.ResponseWriter, req *http.Request) { writeIMRelationError(wrt, req, err) return } + if _, err := store.GroupBusiness.CreateSettings(principal.TenantID, &types.GroupSettings{ + Topic: topicName, + OwnerUserID: principal.Uid, + Name: name, + AvatarFileID: strings.TrimSpace(body.AvatarFileID), + GroupStatus: types.GroupStatusActive, + MemberUploadFiles: true, + MemberPrivateChat: true, + MemberInvite: true, + MaxAdminCount: groupMaxAdminCount, + MemberCount: 1, + Version: 1, + CreatedBy: principal.Uid, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + writeIMRelationError(wrt, req, err) + return + } + if _, err := store.GroupBusiness.UpsertMember(principal.TenantID, &types.GroupMember{ + Topic: topicName, + UserID: principal.Uid, + Role: types.GroupMemberRoleOwner, + MemberStatus: types.GroupMemberStatusActive, + JoinSource: "create", + JoinedBy: principal.Uid, + JoinedAt: now, + Version: 1, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + writeIMRelationError(wrt, req, err) + return + } var members []*types.Subscription seen := map[types.Uid]bool{principal.Uid: true} for _, rawID := range body.MemberUserIDs { @@ -2978,6 +3012,30 @@ func serveIMGroupCreate(wrt http.ResponseWriter, req *http.Request) { writeIMRelationError(wrt, req, err) return } + for _, sub := range members { + uid := types.ParseUid(sub.User) + if uid.IsZero() { + uid = types.ParseUserId(sub.User) + } + if uid.IsZero() { + continue + } + if _, err := store.GroupBusiness.UpsertMember(principal.TenantID, &types.GroupMember{ + Topic: topicName, + UserID: uid, + Role: types.GroupMemberRoleMember, + MemberStatus: types.GroupMemberStatusActive, + JoinSource: "create", + JoinedBy: principal.Uid, + JoinedAt: now, + Version: 1, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + writeIMRelationError(wrt, req, err) + return + } + } _ = store.Topics.UpdateSubCnt(principal.TenantID, topicName) } created, err := store.Topics.Get(principal.TenantID, topicName) @@ -3186,7 +3244,11 @@ func serveIMGroupDetail(wrt http.ResponseWriter, req *http.Request, topic string writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "group not found", nil) return } - writeIMSuccess(wrt, req, groupDetailPayload(group, selfSub)) + payload := groupDetailPayload(group, selfSub) + if settings, err := store.GroupBusiness.GetSettings(principal.TenantID, topic); err == nil { + payload = mergeGroupSettingsPayload(payload, settings) + } + writeIMSuccess(wrt, req, payload) case http.MethodPatch: access := selfSub.ModeGiven & selfSub.ModeWant if !access.IsAdmin() { @@ -3239,12 +3301,18 @@ func serveIMGroupDetail(wrt http.ResponseWriter, req *http.Request, topic string writeIMRelationError(wrt, req, err) return } + patch := types.GroupSettingsPatch{Name: body.Name, AvatarFileID: body.AvatarFileID, Description: body.Description, Version: body.Version} + _, _ = store.GroupBusiness.UpdateSettings(principal.TenantID, topic, patch) updated, err := store.Topics.Get(principal.TenantID, topic) if err != nil { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return } - writeIMSuccess(wrt, req, groupDetailPayload(updated, selfSub)) + payload := groupDetailPayload(updated, selfSub) + if settings, err := store.GroupBusiness.GetSettings(principal.TenantID, topic); err == nil { + payload = mergeGroupSettingsPayload(payload, settings) + } + writeIMSuccess(wrt, req, payload) default: writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) } @@ -3264,7 +3332,12 @@ func serveIMGroupSettings(wrt http.ResponseWriter, req *http.Request, topic stri writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return } - writeIMSuccess(wrt, req, groupSettingsPayload(group, selfSub)) + payload := groupSettingsPayload(group, selfSub) + if settings, err := store.GroupBusiness.GetSettings(principal.TenantID, topic); err == nil { + payload = mergeGroupSettingsPayload(payload, settings) + payload["governance"] = groupGovernancePayloadFromSettings(settings, selfSub) + } + writeIMSuccess(wrt, req, payload) } func serveIMGroupMemberSettings(wrt http.ResponseWriter, req *http.Request, topic string) { @@ -3293,6 +3366,7 @@ func serveIMGroupMemberSettings(wrt http.ResponseWriter, req *http.Request, topi writeIMRelationError(wrt, req, err) return } + _, _ = store.GroupBusiness.UpdateMember(principal.TenantID, topic, principal.Uid, types.GroupMemberPatch{Pinned: body.Pinned, MuteNotifications: body.MuteNotifications}) selfSub.Private = private writeIMSuccess(wrt, req, groupMemberSettingsPayload(selfSub)) } @@ -3332,6 +3406,7 @@ func serveIMGroupMyProfile(wrt http.ResponseWriter, req *http.Request, topic str writeIMRelationError(wrt, req, err) return } + _, _ = store.GroupBusiness.UpdateMember(principal.TenantID, topic, principal.Uid, types.GroupMemberPatch{Nickname: &nickname, Version: body.Version}) writeIMSuccess(wrt, req, map[string]any{"nickname": nickname, "version": currentVersion + 1, "updated_at": updatedAt.Format(time.RFC3339Nano)}) } @@ -3352,6 +3427,10 @@ func serveIMGroupLeave(wrt http.ResponseWriter, req *http.Request, topic string) writeIMRelationError(wrt, req, err) return } + leftAt := time.Now().UTC().Round(time.Millisecond) + leftStatus := types.GroupMemberStatusLeft + _, _ = store.GroupBusiness.UpdateMember(principal.TenantID, topic, principal.Uid, types.GroupMemberPatch{MemberStatus: &leftStatus, LeftAt: &leftAt}) + _, _ = store.GroupBusiness.UpdateSettings(principal.TenantID, topic, types.GroupSettingsPatch{MemberCountDelta: -1}) _ = store.Topics.UpdateSubCnt(principal.TenantID, topic) writeIMSuccess(wrt, req, map[string]any{"topic": topic, "status": "left"}) } @@ -3448,6 +3527,43 @@ func serveIMGroupInvitations(wrt http.ResponseWriter, req *http.Request, topic s item["status"] = "failed" item["reason"] = relationErrorReason(err) } else { + now := time.Now().UTC().Round(time.Millisecond) + if _, err := store.GroupBusiness.CreateInvitation(principal.TenantID, &types.GroupInvitation{ + InvitationID: "ginv_" + store.Store.GetUidString(), + Topic: topic, + InviterUserID: principal.Uid, + TargetUserID: uid, + Message: strings.TrimSpace(body.Message), + Status: types.GroupInvitationStatusJoined, + Result: "joined", + ExpiresAt: now.Add(7 * 24 * time.Hour), + HandledAt: &now, + CreatedAt: now, + UpdatedAt: now, + IdempotencyKey: strings.TrimSpace(req.Header.Get("Idempotency-Key")), + }); err != nil { + item["status"] = "failed" + item["reason"] = relationErrorReason(err) + results = append(results, item) + continue + } + if _, err := store.GroupBusiness.UpsertMember(principal.TenantID, &types.GroupMember{ + Topic: topic, + UserID: uid, + Role: types.GroupMemberRoleMember, + MemberStatus: types.GroupMemberStatusActive, + JoinSource: "invitation", + JoinedBy: principal.Uid, + JoinedAt: now, + Version: 1, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + item["status"] = "failed" + item["reason"] = relationErrorReason(err) + results = append(results, item) + continue + } item["status"] = "joined" } results = append(results, item) @@ -3467,50 +3583,42 @@ func serveIMGroupJoinRequests(wrt http.ResponseWriter, req *http.Request, topic writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil) return } - group, err := store.Topics.Get(principal.TenantID, topic) - if err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return - } status := strings.TrimSpace(req.URL.Query().Get("status")) if status == "" { status = "pending" } - items := filterGroupJoinRequests(groupJoinRequests(group), status) - writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": items, "next_cursor": ""}) + items, nextCursor, err := store.GroupBusiness.ListJoinRequests(principal.TenantID, topic, status, strings.TrimSpace(req.URL.Query().Get("cursor")), parseIMLimit(req.URL.Query().Get("limit"), 50, 100)) + if err != nil { + if !errors.Is(err, types.ErrNotFound) { + writeIMRelationError(wrt, req, err) + return + } + items = nil + } + writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": groupJoinRequestPayloads(items), "next_cursor": nextCursor}) case http.MethodPost: var body imGroupJoinRequestCreateRequest if !decodeIMJSONBody(wrt, req, &body) { return } - group, err := store.Topics.Get(principal.TenantID, topic) - if err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return - } - aux := groupAuxMap(group.Aux) - requests := groupJoinRequests(group) now := time.Now().UTC().Round(time.Millisecond) requestID := "gjr_" + store.Store.GetUidString() - item := map[string]any{ - "request_id": requestID, - "user_id": principal.Uid.UserId(), - "inviter_id": "", - "source": defaultString(strings.TrimSpace(body.Source), "manual"), - "message": strings.TrimSpace(body.Message), - "status": "pending", - "version": 1, - "created_at": now.Format(time.RFC3339Nano), - "updated_at": now.Format(time.RFC3339Nano), - "qr_token": strings.TrimSpace(body.QRToken), - } - requests = append(requests, item) - aux["join_requests"] = requests - if err = store.Topics.Update(principal.TenantID, topic, map[string]any{"Aux": aux}); err != nil { + item, err := store.GroupBusiness.CreateJoinRequest(principal.TenantID, &types.GroupJoinRequest{ + RequestID: requestID, + Topic: topic, + ApplicantUserID: principal.Uid, + Source: defaultString(strings.TrimSpace(body.Source), "manual"), + Message: strings.TrimSpace(body.Message), + Status: types.GroupJoinRequestStatusPending, + Version: 1, + CreatedAt: now, + UpdatedAt: now, + }) + if err != nil { writeIMRelationError(wrt, req, err) return } - writeIMSuccess(wrt, req, item) + writeIMSuccess(wrt, req, groupJoinRequestPayload(item)) default: writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) } @@ -3538,25 +3646,13 @@ func serveIMGroupJoinRequestAction(wrt http.ResponseWriter, req *http.Request, t if !decodeIMJSONBody(wrt, req, &body) { return } - group, err := store.Topics.Get(principal.TenantID, topic) + updated, err := store.GroupBusiness.ResolveJoinRequest(principal.TenantID, topic, requestID, principal.Uid, action, "", body.Version, time.Now().UTC().Round(time.Millisecond)) if err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return - } - aux := groupAuxMap(group.Aux) - requests := groupJoinRequests(group) - updated, found := resolveGroupJoinRequest(requests, requestID, action, principal.Uid, body.Version) - if !found { - writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "join request not found", nil) - return - } - aux["join_requests"] = requests - if err = store.Topics.Update(principal.TenantID, topic, map[string]any{"Aux": aux}); err != nil { writeIMRelationError(wrt, req, err) return } if action == "accept" { - uid := types.ParseUserId(stringFromAny(updated["user_id"])) + uid := updated.ApplicantUserID if !uid.IsZero() { _ = store.Subs.Create(principal.TenantID, &types.Subscription{ TenantID: principal.TenantID, @@ -3566,10 +3662,12 @@ func serveIMGroupJoinRequestAction(wrt http.ResponseWriter, req *http.Request, t ModeGiven: types.ModeCPublic, Private: groupMemberPrivate("", false, false, 1, time.Now().UTC().Round(time.Millisecond)), }) + now := time.Now().UTC().Round(time.Millisecond) + _, _ = store.GroupBusiness.UpsertMember(principal.TenantID, &types.GroupMember{Topic: topic, UserID: uid, Role: types.GroupMemberRoleMember, MemberStatus: types.GroupMemberStatusActive, JoinSource: "approval", JoinedBy: principal.Uid, JoinedAt: now, Version: 1, CreatedAt: now, UpdatedAt: now}) _ = store.Topics.UpdateSubCnt(principal.TenantID, topic) } } - writeIMSuccess(wrt, req, updated) + writeIMSuccess(wrt, req, groupJoinRequestPayload(updated)) } func serveIMGroupSummary(wrt http.ResponseWriter, req *http.Request, topic string) { @@ -3581,16 +3679,12 @@ func serveIMGroupSummary(wrt http.ResponseWriter, req *http.Request, topic strin if !ok { return } - group, err := store.Topics.Get(principal.TenantID, topic) - if err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return - } - announcement := latestGroupAnnouncement(group) + requests, _, _ := store.GroupBusiness.ListJoinRequests(principal.TenantID, topic, types.GroupJoinRequestStatusPending, "", 100) + announcement, _ := store.GroupBusiness.LatestAnnouncement(principal.TenantID, topic) writeIMSuccess(wrt, req, map[string]any{ "topic": topic, - "pending_join_request_count": len(filterGroupJoinRequests(groupJoinRequests(group), "pending")), - "announcement": announcement, + "pending_join_request_count": len(requests), + "announcement": groupAnnouncementPayload(announcement), "permissions": groupPermissions(selfSub.ModeGiven & selfSub.ModeWant), }) } @@ -3600,14 +3694,14 @@ func serveIMGroupAnnouncements(wrt http.ResponseWriter, req *http.Request, topic if !ok { return } - group, err := store.Topics.Get(principal.TenantID, topic) - if err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return - } switch { case tail == "announcements/latest" && req.Method == http.MethodGet: - writeIMSuccess(wrt, req, latestGroupAnnouncement(group)) + ann, err := store.GroupBusiness.LatestAnnouncement(principal.TenantID, topic) + if err != nil && !errors.Is(err, types.ErrNotFound) { + writeIMRelationError(wrt, req, err) + return + } + writeIMSuccess(wrt, req, groupAnnouncementPayload(ann)) case tail == "announcements/current" && req.Method == http.MethodPut: if !(selfSub.ModeGiven & selfSub.ModeWant).IsAdmin() { writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil) @@ -3622,36 +3716,35 @@ func serveIMGroupAnnouncements(wrt http.ResponseWriter, req *http.Request, topic writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "content"}) return } - aux := groupAuxMap(group.Aux) - current := latestGroupAnnouncement(group) - version := intFromAny(current["version"]) - if version <= 0 { - version = 1 - } - if body.Version > 0 && body.Version != version { - writeIMError(wrt, req, http.StatusConflict, 30006, "VERSION_CONFLICT", "version conflict", map[string]any{"current_version": version}) - return - } now := time.Now().UTC().Round(time.Millisecond) - announcement := map[string]any{ - "announcement_id": defaultString(stringFromAny(current["announcement_id"]), "ann_"+store.Store.GetUidString()), - "content": content, - "author": principal.Uid.UserId(), - "version": version + 1, - "published_at": now.Format(time.RFC3339Nano), - } - aux["announcement"] = announcement - if err = store.Topics.Update(principal.TenantID, topic, map[string]any{"Aux": aux}); err != nil { + announcement, err := store.GroupBusiness.PutAnnouncement(principal.TenantID, &types.GroupAnnouncement{ + AnnouncementID: "ann_" + store.Store.GetUidString(), + Topic: topic, + AuthorUserID: principal.Uid, + Content: content, + Status: types.GroupAnnouncementStatusCurrent, + PublishedAt: now, + }, body.Version) + if err != nil { writeIMRelationError(wrt, req, err) return } - writeIMSuccess(wrt, req, announcement) + writeIMSuccess(wrt, req, groupAnnouncementPayload(announcement)) case strings.HasPrefix(tail, "announcements/") && strings.HasSuffix(tail, "/read") && req.Method == http.MethodPost: announcementID := strings.TrimSuffix(strings.TrimPrefix(tail, "announcements/"), "/read") now := time.Now().UTC().Round(time.Millisecond) + if err := store.GroupBusiness.ReadAnnouncement(principal.TenantID, topic, announcementID, principal.Uid, now); err != nil { + writeIMRelationError(wrt, req, err) + return + } writeIMSuccess(wrt, req, map[string]any{"announcement_id": announcementID, "read_at": now.Format(time.RFC3339Nano)}) case strings.HasPrefix(tail, "announcements/") && req.Method == http.MethodGet: - writeIMSuccess(wrt, req, latestGroupAnnouncement(group)) + ann, err := store.GroupBusiness.LatestAnnouncement(principal.TenantID, topic) + if err != nil && !errors.Is(err, types.ErrNotFound) { + writeIMRelationError(wrt, req, err) + return + } + writeIMSuccess(wrt, req, groupAnnouncementPayload(ann)) default: writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) } @@ -3662,13 +3755,18 @@ func serveIMGroupGovernance(wrt http.ResponseWriter, req *http.Request, topic st if !ok { return } - group, err := store.Topics.Get(principal.TenantID, topic) - if err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return - } switch req.Method { case http.MethodGet: + settings, err := store.GroupBusiness.GetSettings(principal.TenantID, topic) + if err != nil && !errors.Is(err, types.ErrNotFound) { + writeIMRelationError(wrt, req, err) + return + } + if settings != nil { + writeIMSuccess(wrt, req, groupGovernancePayloadFromSettings(settings, selfSub)) + return + } + group, _ := store.Topics.Get(principal.TenantID, topic) writeIMSuccess(wrt, req, groupGovernancePayload(group, selfSub)) case http.MethodPatch: if !(selfSub.ModeGiven & selfSub.ModeWant).IsAdmin() { @@ -3679,28 +3777,19 @@ func serveIMGroupGovernance(wrt http.ResponseWriter, req *http.Request, topic st if !decodeIMJSONBody(wrt, req, &body) { return } - aux := groupAuxMap(group.Aux) - governance := groupGovernanceMap(aux["governance"]) - currentVersion := intFromAny(governance["version"]) - if currentVersion <= 0 { - currentVersion = 1 - } - if body.Version > 0 && body.Version != currentVersion { - writeIMError(wrt, req, http.StatusConflict, 30006, "VERSION_CONFLICT", "version conflict", map[string]any{"current_version": currentVersion}) - return - } - applyBoolPatch(governance, "join_approval", body.JoinApproval) - applyBoolPatch(governance, "mute_all", body.MuteAll) - applyBoolPatch(governance, "member_upload_files", body.MemberUploadFiles) - applyBoolPatch(governance, "member_private_chat", body.MemberPrivateChat) - applyBoolPatch(governance, "member_invite", body.MemberInvite) - governance["version"] = currentVersion + 1 - aux["governance"] = governance - if err = store.Topics.Update(principal.TenantID, topic, map[string]any{"Aux": aux}); err != nil { + settings, err := store.GroupBusiness.UpdateSettings(principal.TenantID, topic, types.GroupSettingsPatch{ + JoinApproval: body.JoinApproval, + MuteAll: body.MuteAll, + MemberUploadFiles: body.MemberUploadFiles, + MemberPrivateChat: body.MemberPrivateChat, + MemberInvite: body.MemberInvite, + Version: body.Version, + }) + if err != nil { writeIMRelationError(wrt, req, err) return } - writeIMSuccess(wrt, req, governance) + writeIMSuccess(wrt, req, groupGovernancePayloadFromSettings(settings, selfSub)) default: writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) } @@ -3770,6 +3859,8 @@ func serveIMGroupAdmins(wrt http.ResponseWriter, req *http.Request, topic, tail item["ok"] = false item["reason"] = relationErrorReason(err) } else { + role := types.GroupMemberRoleAdmin + _, _ = store.GroupBusiness.UpdateMember(principal.TenantID, topic, uid, types.GroupMemberPatch{Role: &role}) item["ok"] = true } results = append(results, item) @@ -3787,6 +3878,8 @@ func serveIMGroupAdmins(wrt http.ResponseWriter, req *http.Request, topic, tail writeIMRelationError(wrt, req, err) return } + role := types.GroupMemberRoleMember + _, _ = store.GroupBusiness.UpdateMember(principal.TenantID, topic, uid, types.GroupMemberPatch{Role: &role}) writeIMSuccess(wrt, req, map[string]any{"user_id": uid.UserId(), "role": "member", "updated_at": time.Now().UTC().Format(time.RFC3339Nano)}) default: writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) @@ -3829,6 +3922,11 @@ func serveIMGroupOwnerTransfer(wrt http.ResponseWriter, req *http.Request, topic return } _ = store.Topics.OwnerChange(principal.TenantID, topic, target) + adminRole := types.GroupMemberRoleAdmin + ownerRole := types.GroupMemberRoleOwner + _, _ = store.GroupBusiness.UpdateMember(principal.TenantID, topic, principal.Uid, types.GroupMemberPatch{Role: &adminRole}) + _, _ = store.GroupBusiness.UpdateMember(principal.TenantID, topic, target, types.GroupMemberPatch{Role: &ownerRole}) + _, _ = store.GroupBusiness.UpdateSettings(principal.TenantID, topic, types.GroupSettingsPatch{OwnerUserID: &target, Version: body.Version}) writeIMSuccess(wrt, req, map[string]any{"owner": target.UserId(), "previous_owner": principal.Uid.UserId(), "version": body.Version + 1}) } @@ -3841,15 +3939,14 @@ func serveIMGroupMutes(wrt http.ResponseWriter, req *http.Request, topic, tail s writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil) return } - group, err := store.Topics.Get(principal.TenantID, topic) - if err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return - } - aux := groupAuxMap(group.Aux) switch { case tail == "mutes" && req.Method == http.MethodGet: - writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": groupMuteItems(aux), "next_cursor": ""}) + items, nextCursor, err := store.GroupBusiness.ListMutes(principal.TenantID, topic, strings.TrimSpace(req.URL.Query().Get("cursor")), parseIMLimit(req.URL.Query().Get("limit"), 50, 100)) + if err != nil && !errors.Is(err, types.ErrNotFound) { + writeIMRelationError(wrt, req, err) + return + } + writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": groupMutePayloads(items), "next_cursor": nextCursor}) case tail == "mutes" && req.Method == http.MethodPost: var body imGroupMuteBatchRequest if !decodeIMJSONBody(wrt, req, &body) { @@ -3859,29 +3956,34 @@ func serveIMGroupMutes(wrt http.ResponseWriter, req *http.Request, topic, tail s writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "members"}) return } - mutes := groupMuteItems(aux) now := time.Now().UTC().Round(time.Millisecond) + mutes := make([]types.GroupMute, 0, len(body.Members)) for _, member := range body.Members { uid := types.ParseUserId(strings.TrimSpace(member.UserID)) if uid.IsZero() { continue } - mutes = upsertGroupMute(mutes, map[string]any{"user_id": uid.UserId(), "muted_by": principal.Uid.UserId(), "muted_at": now.Format(time.RFC3339Nano), "expires_at": strings.TrimSpace(member.ExpiresAt), "reason": strings.TrimSpace(body.Reason)}) + var expiresAt *time.Time + if raw := strings.TrimSpace(member.ExpiresAt); raw != "" { + if parsed, err := time.Parse(time.RFC3339Nano, raw); err == nil { + expiresAt = &parsed + } + } + mute, err := store.GroupBusiness.UpsertMute(principal.TenantID, &types.GroupMute{Topic: topic, UserID: uid, MutedBy: principal.Uid, MutedAt: now, ExpiresAt: expiresAt, Reason: strings.TrimSpace(body.Reason), Status: types.GroupMuteStatusActive}) + if err != nil { + writeIMRelationError(wrt, req, err) + return + } + mutes = append(mutes, *mute) } - aux["mutes"] = mutes - if err = store.Topics.Update(principal.TenantID, topic, map[string]any{"Aux": aux}); err != nil { - writeIMRelationError(wrt, req, err) - return - } - writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": mutes}) + writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": groupMutePayloads(mutes)}) case strings.HasPrefix(tail, "mutes/") && req.Method == http.MethodDelete: uid := types.ParseUserId(strings.TrimSpace(strings.TrimPrefix(tail, "mutes/"))) if uid.IsZero() { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "user_id"}) return } - aux["mutes"] = removeGroupMute(groupMuteItems(aux), uid.UserId()) - if err = store.Topics.Update(principal.TenantID, topic, map[string]any{"Aux": aux}); err != nil { + if err := store.GroupBusiness.DeleteMute(principal.TenantID, topic, uid, principal.Uid, time.Now().UTC().Round(time.Millisecond)); err != nil { writeIMRelationError(wrt, req, err) return } @@ -4919,6 +5021,114 @@ func groupGovernancePayload(group *types.Topic, selfSub *types.Subscription) map return governance } +func groupGovernancePayloadFromSettings(settings *types.GroupSettings, selfSub *types.Subscription) map[string]any { + governance := defaultGroupGovernance() + if settings != nil { + governance["join_approval"] = settings.JoinApproval + governance["mute_all"] = settings.MuteAll + governance["member_upload_files"] = settings.MemberUploadFiles + governance["member_private_chat"] = settings.MemberPrivateChat + governance["member_invite"] = settings.MemberInvite + governance["version"] = settings.Version + } + editable := false + if selfSub != nil { + editable = (selfSub.ModeGiven & selfSub.ModeWant).IsAdmin() + } + for _, key := range []string{"join_approval", "mute_all", "member_upload_files", "member_private_chat", "member_invite"} { + governance[key+"_editable"] = editable + } + return governance +} + +func mergeGroupSettingsPayload(payload map[string]any, settings *types.GroupSettings) map[string]any { + if payload == nil { + payload = map[string]any{} + } + if settings == nil { + return payload + } + payload["topic"] = settings.Topic + payload["name"] = settings.Name + payload["avatar_file_id"] = settings.AvatarFileID + payload["description"] = settings.Description + payload["owner"] = settings.OwnerUserID.UserId() + payload["member_count"] = settings.MemberCount + payload["version"] = settings.Version + return payload +} + +func groupJoinRequestPayload(req *types.GroupJoinRequest) map[string]any { + if req == nil { + return map[string]any{} + } + data := map[string]any{ + "request_id": req.RequestID, + "user_id": req.ApplicantUserID.UserId(), + "inviter_id": req.InviterUserID.UserId(), + "source": req.Source, + "message": req.Message, + "status": req.Status, + "version": req.Version, + "created_at": req.CreatedAt.UTC().Format(time.RFC3339Nano), + "updated_at": req.UpdatedAt.UTC().Format(time.RFC3339Nano), + } + if req.HandledBy != types.ZeroUid { + data["handled_by"] = req.HandledBy.UserId() + } + if req.HandledAt != nil { + data["handled_at"] = req.HandledAt.UTC().Format(time.RFC3339Nano) + } + if req.RejectReason != "" { + data["reject_reason"] = req.RejectReason + } + return data +} + +func groupJoinRequestPayloads(items []types.GroupJoinRequest) []map[string]any { + out := make([]map[string]any, 0, len(items)) + for i := range items { + out = append(out, groupJoinRequestPayload(&items[i])) + } + return out +} + +func groupAnnouncementPayload(ann *types.GroupAnnouncement) map[string]any { + if ann == nil { + return map[string]any{} + } + return map[string]any{ + "announcement_id": ann.AnnouncementID, + "content": ann.Content, + "author": ann.AuthorUserID.UserId(), + "version": ann.Version, + "published_at": ann.PublishedAt.UTC().Format(time.RFC3339Nano), + } +} + +func groupMutePayload(mute types.GroupMute) map[string]any { + data := map[string]any{ + "user_id": mute.UserID.UserId(), + "muted_by": mute.MutedBy.UserId(), + "muted_at": mute.MutedAt.UTC().Format(time.RFC3339Nano), + "reason": mute.Reason, + "status": mute.Status, + "version": mute.Version, + } + if mute.ExpiresAt != nil { + data["expires_at"] = mute.ExpiresAt.UTC().Format(time.RFC3339Nano) + } + return data +} + +func groupMutePayloads(items []types.GroupMute) []map[string]any { + out := make([]map[string]any, 0, len(items)) + for _, item := range items { + out = append(out, groupMutePayload(item)) + } + return out +} + func groupAuxMap(value any) map[string]any { switch aux := value.(type) { case map[string]any: diff --git a/server/im_http_test.go b/server/im_http_test.go index 2112c5c..f2a5380 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -138,6 +138,142 @@ type testSafetyStore struct { targetType string } +type testGroupBusinessStore struct { + settings *types.GroupSettings + member *types.GroupMember + joinRequests []types.GroupJoinRequest + announcement *types.GroupAnnouncement + mutes []types.GroupMute + invitations []types.GroupInvitation + updatedSettings types.GroupSettingsPatch + updatedMembers []types.GroupMemberPatch +} + +func (s *testGroupBusinessStore) CreateSettings(_ types.TenantID, settings *types.GroupSettings) (*types.GroupSettings, error) { + s.settings = settings + return settings, nil +} + +func (s *testGroupBusinessStore) GetSettings(_ types.TenantID, topic string) (*types.GroupSettings, error) { + if s.settings != nil { + return s.settings, nil + } + return &types.GroupSettings{Topic: topic, OwnerUserID: types.Uid(42), Name: "Alpha", GroupStatus: types.GroupStatusActive, MemberUploadFiles: true, MemberPrivateChat: true, MemberInvite: true, MaxAdminCount: groupMaxAdminCount, MemberCount: 3, Version: 1}, nil +} + +func (s *testGroupBusinessStore) UpdateSettings(_ types.TenantID, topic string, patch types.GroupSettingsPatch) (*types.GroupSettings, error) { + s.updatedSettings = patch + settings, _ := s.GetSettings(0, topic) + if patch.Name != nil { + settings.Name = *patch.Name + } + if patch.Description != nil { + settings.Description = *patch.Description + } + if patch.JoinApproval != nil { + settings.JoinApproval = *patch.JoinApproval + } + if patch.OwnerUserID != nil { + settings.OwnerUserID = *patch.OwnerUserID + } + settings.Version++ + s.settings = settings + return settings, nil +} + +func (s *testGroupBusinessStore) UpsertMember(_ types.TenantID, member *types.GroupMember) (*types.GroupMember, error) { + s.member = member + return member, nil +} + +func (s *testGroupBusinessStore) GetMember(_ types.TenantID, topic string, uid types.Uid) (*types.GroupMember, error) { + return &types.GroupMember{Topic: topic, UserID: uid, Role: types.GroupMemberRoleMember, MemberStatus: types.GroupMemberStatusActive, Version: 1}, nil +} + +func (s *testGroupBusinessStore) ListMembers(_ types.TenantID, _ string, _ types.GroupMemberListOptions) ([]types.GroupMember, string, error) { + return nil, "", nil +} + +func (s *testGroupBusinessStore) UpdateMember(_ types.TenantID, topic string, uid types.Uid, patch types.GroupMemberPatch) (*types.GroupMember, error) { + s.updatedMembers = append(s.updatedMembers, patch) + return &types.GroupMember{Topic: topic, UserID: uid, Role: defaultStringPtr(patch.Role, types.GroupMemberRoleMember), MemberStatus: types.GroupMemberStatusActive, Version: 2}, nil +} + +func (s *testGroupBusinessStore) CreateInvitation(_ types.TenantID, inv *types.GroupInvitation) (*types.GroupInvitation, error) { + s.invitations = append(s.invitations, *inv) + return inv, nil +} + +func (s *testGroupBusinessStore) CreateJoinRequest(_ types.TenantID, req *types.GroupJoinRequest) (*types.GroupJoinRequest, error) { + now := time.Now().UTC() + req.CreatedAt = now + req.UpdatedAt = now + s.joinRequests = append(s.joinRequests, *req) + return req, nil +} + +func (s *testGroupBusinessStore) ListJoinRequests(_ types.TenantID, _ string, status, _ string, _ int) ([]types.GroupJoinRequest, string, error) { + if status == "" || status == "all" { + return s.joinRequests, "", nil + } + out := []types.GroupJoinRequest{} + for _, item := range s.joinRequests { + if item.Status == status { + out = append(out, item) + } + } + return out, "", nil +} + +func (s *testGroupBusinessStore) ResolveJoinRequest(_ types.TenantID, topic, requestID string, actor types.Uid, action, _ string, _ int, now time.Time) (*types.GroupJoinRequest, error) { + status := types.GroupJoinRequestStatusRejected + if action == "accept" { + status = types.GroupJoinRequestStatusAccepted + } + return &types.GroupJoinRequest{RequestID: requestID, Topic: topic, ApplicantUserID: types.Uid(43), Status: status, HandledBy: actor, HandledAt: &now, Version: 2, CreatedAt: now, UpdatedAt: now}, nil +} + +func (s *testGroupBusinessStore) LatestAnnouncement(_ types.TenantID, topic string) (*types.GroupAnnouncement, error) { + if s.announcement == nil { + return nil, types.ErrNotFound + } + return s.announcement, nil +} + +func (s *testGroupBusinessStore) PutAnnouncement(_ types.TenantID, ann *types.GroupAnnouncement, _ int) (*types.GroupAnnouncement, error) { + ann.Version = 2 + if ann.PublishedAt.IsZero() { + ann.PublishedAt = time.Now().UTC() + } + s.announcement = ann + return ann, nil +} + +func (s *testGroupBusinessStore) ReadAnnouncement(_ types.TenantID, _, _ string, _ types.Uid, _ time.Time) error { + return nil +} + +func (s *testGroupBusinessStore) ListMutes(_ types.TenantID, _ string, _ string, _ int) ([]types.GroupMute, string, error) { + return s.mutes, "", nil +} + +func (s *testGroupBusinessStore) UpsertMute(_ types.TenantID, mute *types.GroupMute) (*types.GroupMute, error) { + mute.Version = 1 + s.mutes = append(s.mutes, *mute) + return mute, nil +} + +func (s *testGroupBusinessStore) DeleteMute(_ types.TenantID, _ string, _ types.Uid, _ types.Uid, _ time.Time) error { + return nil +} + +func defaultStringPtr(value *string, fallback string) string { + if value == nil { + return fallback + } + return *value +} + func (s *testWorkbenchStore) Get(tenantID types.TenantID, uid types.Uid) ([]types.WorkbenchModule, int, error) { s.tenantID = tenantID s.uid = uid @@ -2167,6 +2303,7 @@ func TestServeIMGroupCreate(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() oldStore := store.Store + oldGroupBusiness := store.GroupBusiness mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) tokenAuth := mock_auth.NewMockAuthHandler(ctrl) mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth) @@ -2181,10 +2318,12 @@ func TestServeIMGroupCreate(t *testing.T) { subs := mock_store.NewMockSubsPersistenceInterface(ctrl) store.Topics = topics store.Subs = subs + store.GroupBusiness = &testGroupBusinessStore{} defer func() { store.Store = oldStore store.Topics = oldTopics store.Subs = oldSubs + store.GroupBusiness = oldGroupBusiness }() topics.EXPECT().Create(types.TenantID(7), gomock.Any(), types.Uid(42), gomock.Any()).DoAndReturn( @@ -2230,6 +2369,7 @@ func TestServeIMGroupDetailPatchSettingsProfileAndLeave(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() oldStore := store.Store + oldGroupBusiness := store.GroupBusiness mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) tokenAuth := mock_auth.NewMockAuthHandler(ctrl) mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth).Times(3) @@ -2243,10 +2383,12 @@ func TestServeIMGroupDetailPatchSettingsProfileAndLeave(t *testing.T) { subs := mock_store.NewMockSubsPersistenceInterface(ctrl) store.Topics = topics store.Subs = subs + store.GroupBusiness = &testGroupBusinessStore{} defer func() { store.Store = oldStore store.Topics = oldTopics store.Subs = oldSubs + store.GroupBusiness = oldGroupBusiness }() selfSub := &types.Subscription{ TenantID: 7, @@ -2315,13 +2457,16 @@ func TestServeIMGroupLeave(t *testing.T) { expectIMBearer(t, ctrl, "token-1", 7, 42) oldTopics := store.Topics oldSubs := store.Subs + oldGroupBusiness := store.GroupBusiness topics := mock_store.NewMockTopicsPersistenceInterface(ctrl) subs := mock_store.NewMockSubsPersistenceInterface(ctrl) store.Topics = topics store.Subs = subs + store.GroupBusiness = &testGroupBusinessStore{} defer func() { store.Topics = oldTopics store.Subs = oldSubs + store.GroupBusiness = oldGroupBusiness }() subs.EXPECT().Get(types.TenantID(7), "grpAlpha", types.Uid(42), false).Return(&types.Subscription{ TenantID: 7, @@ -2426,6 +2571,7 @@ func TestServeIMGroupAnnouncementGovernanceJoinRequest(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() oldStore := store.Store + oldGroupBusiness := store.GroupBusiness mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) tokenAuth := mock_auth.NewMockAuthHandler(ctrl) mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth).Times(3) @@ -2441,16 +2587,15 @@ func TestServeIMGroupAnnouncementGovernanceJoinRequest(t *testing.T) { subs := mock_store.NewMockSubsPersistenceInterface(ctrl) store.Topics = topics store.Subs = subs + store.GroupBusiness = &testGroupBusinessStore{} defer func() { store.Store = oldStore store.Topics = oldTopics store.Subs = oldSubs + store.GroupBusiness = oldGroupBusiness }() selfSub := &types.Subscription{TenantID: 7, User: types.Uid(42).String(), Topic: "grpAlpha", ModeWant: types.ModeCFull, ModeGiven: types.ModeCFull} - group := &types.Topic{ObjHeader: types.ObjHeader{Id: "grpAlpha"}, TenantID: 7, Owner: types.Uid(42).String(), Aux: types.KVMap{}} subs.EXPECT().Get(types.TenantID(7), "grpAlpha", types.Uid(42), false).Return(selfSub, nil).Times(3) - topics.EXPECT().Get(types.TenantID(7), "grpAlpha").Return(group, nil).Times(3) - topics.EXPECT().Update(types.TenantID(7), "grpAlpha", gomock.Any()).Return(nil).Times(3) annReq := httptest.NewRequest(http.MethodPut, "/v1/im/groups/grpAlpha/announcements/current", bytes.NewBufferString(`{"content":"公告","version":1}`)) annReq.Header.Set("Authorization", "Bearer token-1") @@ -2490,6 +2635,7 @@ func TestServeIMGroupAdminsMutesAndOwnerTransfer(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() oldStore := store.Store + oldGroupBusiness := store.GroupBusiness mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) tokenAuth := mock_auth.NewMockAuthHandler(ctrl) mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth).Times(3) @@ -2503,22 +2649,21 @@ func TestServeIMGroupAdminsMutesAndOwnerTransfer(t *testing.T) { subs := mock_store.NewMockSubsPersistenceInterface(ctrl) store.Topics = topics store.Subs = subs + store.GroupBusiness = &testGroupBusinessStore{} defer func() { store.Store = oldStore store.Topics = oldTopics store.Subs = oldSubs + store.GroupBusiness = oldGroupBusiness }() owner := types.Uid(42) target := types.Uid(43) ownerSub := &types.Subscription{TenantID: 7, User: owner.String(), Topic: "grpAlpha", ModeWant: types.ModeCFull, ModeGiven: types.ModeCFull} - group := &types.Topic{ObjHeader: types.ObjHeader{Id: "grpAlpha"}, TenantID: 7, Owner: owner.String(), Aux: types.KVMap{}} member := types.Subscription{TenantID: 7, User: target.UserId(), Topic: "grpAlpha", ModeWant: types.ModeCPublic, ModeGiven: types.ModeCPublic} members := []types.Subscription{*ownerSub, member} subs.EXPECT().Get(types.TenantID(7), "grpAlpha", owner, false).Return(ownerSub, nil).Times(3) topics.EXPECT().GetUsers(types.TenantID(7), "grpAlpha", gomock.Any()).Return(members, nil).Times(2) subs.EXPECT().Update(types.TenantID(7), "grpAlpha", target, gomock.Any()).Return(nil).Times(2) - topics.EXPECT().Get(types.TenantID(7), "grpAlpha").Return(group, nil) - topics.EXPECT().Update(types.TenantID(7), "grpAlpha", gomock.Any()).Return(nil) subs.EXPECT().Get(types.TenantID(7), "grpAlpha", target, false).Return(&member, nil) subs.EXPECT().Update(types.TenantID(7), "grpAlpha", owner, gomock.Any()).Return(nil) topics.EXPECT().OwnerChange(types.TenantID(7), "grpAlpha", target).Return(nil) diff --git a/server/store/store.go b/server/store/store.go index d1c28ca..977c8f1 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -1079,6 +1079,226 @@ func (callsMapper) List(tenantID types.TenantID, uid types.Uid, topic, cursor st return tenantAdp.TenantCallList(tenantID, uid, strings.TrimSpace(topic), strings.TrimSpace(cursor), limit) } +type GroupBusinessPersistenceInterface interface { + CreateSettings(types.TenantID, *types.GroupSettings) (*types.GroupSettings, error) + GetSettings(types.TenantID, string) (*types.GroupSettings, error) + UpdateSettings(types.TenantID, string, types.GroupSettingsPatch) (*types.GroupSettings, error) + UpsertMember(types.TenantID, *types.GroupMember) (*types.GroupMember, error) + GetMember(types.TenantID, string, types.Uid) (*types.GroupMember, error) + ListMembers(types.TenantID, string, types.GroupMemberListOptions) ([]types.GroupMember, string, error) + UpdateMember(types.TenantID, string, types.Uid, types.GroupMemberPatch) (*types.GroupMember, error) + CreateInvitation(types.TenantID, *types.GroupInvitation) (*types.GroupInvitation, error) + CreateJoinRequest(types.TenantID, *types.GroupJoinRequest) (*types.GroupJoinRequest, error) + ListJoinRequests(types.TenantID, string, string, string, int) ([]types.GroupJoinRequest, string, error) + ResolveJoinRequest(types.TenantID, string, string, types.Uid, string, string, int, time.Time) (*types.GroupJoinRequest, error) + LatestAnnouncement(types.TenantID, string) (*types.GroupAnnouncement, error) + PutAnnouncement(types.TenantID, *types.GroupAnnouncement, int) (*types.GroupAnnouncement, error) + ReadAnnouncement(types.TenantID, string, string, types.Uid, time.Time) error + ListMutes(types.TenantID, string, string, int) ([]types.GroupMute, string, error) + UpsertMute(types.TenantID, *types.GroupMute) (*types.GroupMute, error) + DeleteMute(types.TenantID, string, types.Uid, types.Uid, time.Time) error +} + +type groupBusinessMapper struct{} + +var GroupBusiness GroupBusinessPersistenceInterface + +func (groupBusinessMapper) CreateSettings(tenantID types.TenantID, settings *types.GroupSettings) (*types.GroupSettings, error) { + if settings == nil || settings.Topic == "" || settings.OwnerUserID.IsZero() || settings.CreatedBy.IsZero() { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupSettingsCreate(tenantID, settings) +} + +func (groupBusinessMapper) GetSettings(tenantID types.TenantID, topic string) (*types.GroupSettings, error) { + if strings.TrimSpace(topic) == "" { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupSettingsGet(tenantID, strings.TrimSpace(topic)) +} + +func (groupBusinessMapper) UpdateSettings(tenantID types.TenantID, topic string, patch types.GroupSettingsPatch) (*types.GroupSettings, error) { + if strings.TrimSpace(topic) == "" { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupSettingsUpdate(tenantID, strings.TrimSpace(topic), patch) +} + +func (groupBusinessMapper) UpsertMember(tenantID types.TenantID, member *types.GroupMember) (*types.GroupMember, error) { + if member == nil || member.Topic == "" || member.UserID.IsZero() { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupMemberUpsert(tenantID, member) +} + +func (groupBusinessMapper) GetMember(tenantID types.TenantID, topic string, uid types.Uid) (*types.GroupMember, error) { + if strings.TrimSpace(topic) == "" || uid.IsZero() { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupMemberGet(tenantID, strings.TrimSpace(topic), uid) +} + +func (groupBusinessMapper) ListMembers(tenantID types.TenantID, topic string, opts types.GroupMemberListOptions) ([]types.GroupMember, string, error) { + if strings.TrimSpace(topic) == "" { + return nil, "", types.ErrMalformed + } + if opts.Limit <= 0 || opts.Limit > 100 { + opts.Limit = 30 + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, "", err + } + return tenantAdp.TenantGroupMemberList(tenantID, strings.TrimSpace(topic), opts) +} + +func (groupBusinessMapper) UpdateMember(tenantID types.TenantID, topic string, uid types.Uid, patch types.GroupMemberPatch) (*types.GroupMember, error) { + if strings.TrimSpace(topic) == "" || uid.IsZero() { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupMemberUpdate(tenantID, strings.TrimSpace(topic), uid, patch) +} + +func (groupBusinessMapper) CreateInvitation(tenantID types.TenantID, invitation *types.GroupInvitation) (*types.GroupInvitation, error) { + if invitation == nil || invitation.Topic == "" || invitation.InviterUserID.IsZero() || invitation.TargetUserID.IsZero() { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupInvitationCreate(tenantID, invitation) +} + +func (groupBusinessMapper) CreateJoinRequest(tenantID types.TenantID, req *types.GroupJoinRequest) (*types.GroupJoinRequest, error) { + if req == nil || req.Topic == "" || req.ApplicantUserID.IsZero() || req.RequestID == "" { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupJoinRequestCreate(tenantID, req) +} + +func (groupBusinessMapper) ListJoinRequests(tenantID types.TenantID, topic, status, cursor string, limit int) ([]types.GroupJoinRequest, string, error) { + if strings.TrimSpace(topic) == "" { + return nil, "", types.ErrMalformed + } + if limit <= 0 || limit > 100 { + limit = 50 + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, "", err + } + return tenantAdp.TenantGroupJoinRequestList(tenantID, strings.TrimSpace(topic), strings.TrimSpace(status), strings.TrimSpace(cursor), limit) +} + +func (groupBusinessMapper) ResolveJoinRequest(tenantID types.TenantID, topic, requestID string, actor types.Uid, action, rejectReason string, version int, now time.Time) (*types.GroupJoinRequest, error) { + if strings.TrimSpace(topic) == "" || strings.TrimSpace(requestID) == "" || actor.IsZero() || now.IsZero() { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupJoinRequestResolve(tenantID, strings.TrimSpace(topic), strings.TrimSpace(requestID), actor, strings.TrimSpace(action), strings.TrimSpace(rejectReason), version, now) +} + +func (groupBusinessMapper) LatestAnnouncement(tenantID types.TenantID, topic string) (*types.GroupAnnouncement, error) { + if strings.TrimSpace(topic) == "" { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupAnnouncementLatest(tenantID, strings.TrimSpace(topic)) +} + +func (groupBusinessMapper) PutAnnouncement(tenantID types.TenantID, ann *types.GroupAnnouncement, expectedVersion int) (*types.GroupAnnouncement, error) { + if ann == nil || ann.Topic == "" || ann.AuthorUserID.IsZero() || ann.Content == "" { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupAnnouncementPut(tenantID, ann, expectedVersion) +} + +func (groupBusinessMapper) ReadAnnouncement(tenantID types.TenantID, topic, announcementID string, uid types.Uid, readAt time.Time) error { + if strings.TrimSpace(topic) == "" || strings.TrimSpace(announcementID) == "" || uid.IsZero() || readAt.IsZero() { + return types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return err + } + return tenantAdp.TenantGroupAnnouncementRead(tenantID, strings.TrimSpace(topic), strings.TrimSpace(announcementID), uid, readAt) +} + +func (groupBusinessMapper) ListMutes(tenantID types.TenantID, topic, cursor string, limit int) ([]types.GroupMute, string, error) { + if strings.TrimSpace(topic) == "" { + return nil, "", types.ErrMalformed + } + if limit <= 0 || limit > 100 { + limit = 50 + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, "", err + } + return tenantAdp.TenantGroupMuteList(tenantID, strings.TrimSpace(topic), strings.TrimSpace(cursor), limit) +} + +func (groupBusinessMapper) UpsertMute(tenantID types.TenantID, mute *types.GroupMute) (*types.GroupMute, error) { + if mute == nil || mute.Topic == "" || mute.UserID.IsZero() || mute.MutedBy.IsZero() { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupMuteUpsert(tenantID, mute) +} + +func (groupBusinessMapper) DeleteMute(tenantID types.TenantID, topic string, uid, actor types.Uid, now time.Time) error { + if strings.TrimSpace(topic) == "" || uid.IsZero() || actor.IsZero() || now.IsZero() { + return types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return err + } + return tenantAdp.TenantGroupMuteDelete(tenantID, strings.TrimSpace(topic), uid, actor, now) +} + type WorkbenchPersistenceInterface interface { Get(tenantID types.TenantID, uid types.Uid) ([]types.WorkbenchModule, int, error) } @@ -2861,6 +3081,7 @@ func init() { Subs = subsMapper{} Messages = messagesMapper{} Calls = callsMapper{} + GroupBusiness = groupBusinessMapper{} Workbench = workbenchMapper{} Organization = organizationMapper{} Meetings = meetingsMapper{} diff --git a/server/store/types/groups.go b/server/store/types/groups.go new file mode 100644 index 0000000..3f91a0b --- /dev/null +++ b/server/store/types/groups.go @@ -0,0 +1,182 @@ +package types + +import "time" + +const ( + GroupStatusActive = "active" + + GroupMemberRoleOwner = "owner" + GroupMemberRoleAdmin = "admin" + GroupMemberRoleMember = "member" + + GroupMemberStatusActive = "active" + GroupMemberStatusLeft = "left" + + GroupInvitationStatusJoined = "joined" + + GroupJoinRequestStatusPending = "pending" + GroupJoinRequestStatusAccepted = "accepted" + GroupJoinRequestStatusRejected = "rejected" + + GroupAnnouncementStatusCurrent = "current" + GroupAnnouncementStatusSuperseded = "superseded" + + GroupMuteStatusActive = "active" + GroupMuteStatusRevoked = "revoked" +) + +type GroupSettings struct { + Topic string `db:"topic"` + OwnerUserID Uid `db:"-"` + Name string `db:"name"` + AvatarFileID string `db:"avatar_file_id"` + Description string `db:"description"` + GroupStatus string `db:"group_status"` + JoinApproval bool `db:"join_approval"` + MuteAll bool `db:"mute_all"` + MemberUploadFiles bool `db:"member_upload_files"` + MemberPrivateChat bool `db:"member_private_chat"` + MemberInvite bool `db:"member_invite"` + MaxMemberCount int `db:"max_member_count"` + MaxAdminCount int `db:"max_admin_count"` + MemberCount int `db:"member_count"` + Version int `db:"version"` + CreatedBy Uid `db:"-"` + DissolvedBy Uid `db:"-"` + DissolvedAt *time.Time `db:"dissolved_at"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` + OwnerUserIDRaw int64 `db:"owner_user_id"` + CreatedByRaw int64 `db:"created_by"` + DissolvedByRaw *int64 `db:"dissolved_by"` +} + +type GroupSettingsPatch struct { + Name *string + AvatarFileID *string + Description *string + OwnerUserID *Uid + JoinApproval *bool + MuteAll *bool + MemberUploadFiles *bool + MemberPrivateChat *bool + MemberInvite *bool + MemberCountDelta int + Version int +} + +type GroupMember struct { + ID int64 `db:"id"` + Topic string `db:"topic"` + UserID Uid `db:"-"` + Role string `db:"role"` + MemberStatus string `db:"member_status"` + Nickname string `db:"nickname"` + Pinned bool `db:"pinned"` + MuteNotifications bool `db:"mute_notifications"` + JoinSource string `db:"join_source"` + JoinedBy Uid `db:"-"` + JoinedAt time.Time `db:"joined_at"` + LeftAt *time.Time `db:"left_at"` + RemovedBy Uid `db:"-"` + Version int `db:"version"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` + UserIDRaw int64 `db:"user_id"` + JoinedByRaw *int64 `db:"joined_by"` + RemovedByRaw *int64 `db:"removed_by"` +} + +type GroupMemberPatch struct { + Role *string + MemberStatus *string + Nickname *string + Pinned *bool + MuteNotifications *bool + LeftAt *time.Time + RemovedBy *Uid + Version int +} + +type GroupMemberListOptions struct { + Role string + EligibleRole string + Query string + Cursor string + Limit int +} + +type GroupInvitation struct { + ID int64 `db:"id"` + InvitationID string `db:"invitation_id"` + Topic string `db:"topic"` + InviterUserID Uid `db:"-"` + TargetUserID Uid `db:"-"` + Message string `db:"message"` + Status string `db:"status"` + Result string `db:"result"` + JoinRequestID string `db:"join_request_id"` + IdempotencyKey string `db:"idempotency_key"` + ExpiresAt time.Time `db:"expires_at"` + HandledAt *time.Time `db:"handled_at"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` + InviterUserRaw int64 `db:"inviter_user_id"` + TargetUserRaw int64 `db:"target_user_id"` +} + +type GroupJoinRequest struct { + ID int64 `db:"id"` + RequestID string `db:"request_id"` + Topic string `db:"topic"` + ApplicantUserID Uid `db:"-"` + InviterUserID Uid `db:"-"` + Source string `db:"source"` + QRTokenHash string `db:"qr_token_hash"` + Message string `db:"message"` + Status string `db:"status"` + RejectReason string `db:"reject_reason"` + HandledBy Uid `db:"-"` + HandledAt *time.Time `db:"handled_at"` + ExpiresAt *time.Time `db:"expires_at"` + IdempotencyKey string `db:"idempotency_key"` + Version int `db:"version"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` + ApplicantUserRaw int64 `db:"applicant_user_id"` + InviterUserRaw *int64 `db:"inviter_user_id"` + HandledByRaw *int64 `db:"handled_by"` +} + +type GroupAnnouncement struct { + ID int64 `db:"id"` + AnnouncementID string `db:"announcement_id"` + Topic string `db:"topic"` + AuthorUserID Uid `db:"-"` + Content string `db:"content"` + Status string `db:"status"` + Version int `db:"version"` + PublishedAt time.Time `db:"published_at"` + IdempotencyKey string `db:"idempotency_key"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` + AuthorUserIDRaw int64 `db:"author_user_id"` +} + +type GroupMute struct { + Topic string `db:"topic"` + UserID Uid `db:"-"` + MutedBy Uid `db:"-"` + MutedAt time.Time `db:"muted_at"` + ExpiresAt *time.Time `db:"expires_at"` + Reason string `db:"reason"` + Status string `db:"status"` + RevokedBy Uid `db:"-"` + RevokedAt *time.Time `db:"revoked_at"` + Version int `db:"version"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` + UserIDRaw int64 `db:"user_id"` + MutedByRaw int64 `db:"muted_by"` + RevokedRaw *int64 `db:"revoked_by"` +}