feat: persist group business state in tables

2026-08-17 09:50:31 +08:00
parent d8d1639d5b
commit ddf0f7a55e
10 changed files with 2065 additions and 130 deletions
@@ -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)
@@ -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(&currentVersion)
+	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)
+	}
+}