im_server/server/db/mysql/tenant_groups.go

//go:build mysql
// +build mysql

package mysql

import (
	"context"
	"database/sql"
	"strconv"
	"strings"
	"time"

	"github.com/jmoiron/sqlx"
	"github.com/tinode/chat/server/db/common"
	"github.com/tinode/chat/server/store"
	t "github.com/tinode/chat/server/store/types"
)

const (
	groupSettingsSelectColumns = `tenant_id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,
		member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,
		created_by,dissolved_by,dissolved_at,created_at,updated_at`
	groupAnnouncementSelectColumnsBase = `tenant_id,id,announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at,
		(SELECT COUNT(*) FROM im_group_announcement_reads AS ar WHERE ar.tenant_id=im_group_announcements.tenant_id AND ar.topic=im_group_announcements.topic AND ar.announcement_id=im_group_announcements.announcement_id) AS read_count,
		(SELECT COUNT(*) FROM im_group_members AS gm WHERE gm.tenant_id=im_group_announcements.tenant_id AND gm.topic=im_group_announcements.topic AND gm.member_status='active') AS member_count`
	groupMemberSelectColumns = `id,topic,user_id,role,member_status,nickname,pinned,mute_notifications,
		join_source,joined_by,joined_at,left_at,removed_by,version,created_at,updated_at`
	groupMuteSelectColumns = `tenant_id,topic,user_id,muted_by,muted_at,expires_at,reason,status,revoked_by,revoked_at,version,created_at,updated_at`
)

func groupAnnouncementSelectColumns(tenantID t.TenantID) string {
	return tenantShardSQL(tenantID, groupAnnouncementSelectColumnsBase)
}

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()
	}
	settingsID, err := nextSnowflakeID()
	if err != nil {
		return nil, err
	}
	_, err = a.db.ExecContext(ctx, `INSERT INTO im_group_settings
		(id,tenant_id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,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)`,
		settingsID, id, 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, id, 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.MaxMemberCount != nil {
		set = append(set, "max_member_count=?")
		args = append(args, *patch.MaxMemberCount)
	}
	if patch.MaxAdminCount != nil {
		set = append(set, "max_admin_count=?")
		args = append(args, *patch.MaxAdminCount)
	}
	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, id, topic)
	where := "tenant_id=? AND topic=?"
	if patch.Version > 0 {
		where += " AND version=?"
		args = append(args, patch.Version)
	}
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	res, err := tx.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
	}
	if patch.MuteAll != nil {
		if err = syncGroupMuteAllTx(tx, id, topic, *patch.MuteAll, time.Now().UTC().Round(time.Millisecond)); err != nil {
			return nil, err
		}
	}
	if err = tx.Commit(); err != nil {
		return nil, err
	}
	return a.groupSettingsByTopic(ctx, id, topic)
}

func (a *adapter) TenantGroupDissolve(id t.TenantID, topic string, actor t.Uid, reason string, now time.Time) (*t.GroupSettings, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return nil, err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupDissolve(id, topic, actor, reason, now)
	}
	if strings.TrimSpace(topic) == "" || actor.IsZero() {
		return nil, t.ErrMalformed
	}
	if now.IsZero() {
		now = time.Now().UTC().Round(time.Millisecond)
	}
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	res, err := tx.ExecContext(ctx, `UPDATE im_group_settings
		SET group_status='dissolved',dissolved_by=?,dissolved_at=?,member_count=0,version=version+1,updated_at=?
		WHERE tenant_id=? AND topic=? AND group_status<>'dissolved'`, decodeUID(actor), now, now, id, strings.TrimSpace(topic))
	if err != nil {
		return nil, err
	}
	if affected, _ := res.RowsAffected(); affected == 0 {
		return nil, t.ErrNotFound
	}
	if _, err = tx.ExecContext(ctx, tenantShardSQL(id, `UPDATE im_subscriptions
		SET updatedat=?,deletedat=?
		WHERE tenant_id=? AND topic IN (?,?) AND deletedat IS NULL`), now, now, id, strings.TrimSpace(topic), t.GrpToChn(strings.TrimSpace(topic))); err != nil {
		return nil, err
	}
	if _, err = tx.ExecContext(ctx, tenantShardSQL(id, `UPDATE im_group_members
		SET member_status=?,left_at=?,removed_by=?,version=version+1,updated_at=?
		WHERE tenant_id=? AND topic=? AND member_status=?`), t.GroupMemberStatusLeft, now, decodeUID(actor), now, id, strings.TrimSpace(topic), t.GroupMemberStatusActive); err != nil {
		return nil, err
	}
	if _, err = tx.ExecContext(ctx, tenantShardSQL(id, `UPDATE im_topics
		SET updatedat=?,touchedat=?,state=?,stateat=?,subcnt=0
		WHERE tenant_id=? AND name=?`), now, now, t.StateDeleted, now, id, strings.TrimSpace(topic)); err != nil {
		return nil, err
	}
	if err = tx.Commit(); err != nil {
		return nil, err
	}
	return a.groupSettingsByTopic(ctx, id, 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, tenantShardSQL(id, `INSERT INTO im_group_members
		(tenant_id,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)`,
	), id, 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, id, 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, tenantShardSQL(id, "SELECT "+groupMemberSelectColumns+" FROM im_group_members WHERE tenant_id=? AND topic=? AND user_id=?"), 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
	}
	memberIndex := "idx_im_group_members_topic_status"
	where := []string{"gm.tenant_id=?", "gm.topic=?", "gm.member_status=?"}
	args := []any{id, topic, t.GroupMemberStatusActive}
	if opts.Role != "" {
		memberIndex = "idx_im_group_members_topic_role"
		where = append(where, "gm.role=?")
		args = append(args, opts.Role)
	}
	if opts.Query != "" {
		like := "%" + opts.Query + "%"
		where = append(where, `(gm.nickname LIKE ? OR p.nickname LIKE ? OR p.public_account LIKE ?
			OR u.username LIKE ?)`)
		args = append(args, like, like, like, like)
	}
	if opts.Cursor != "" {
		parsed, err := strconv.ParseInt(opts.Cursor, 10, 64)
		if err != nil || parsed < 0 {
			return nil, "", t.ErrMalformed
		}
		if parsed > 0 {
			where = append(where, "gm.id>?")
			args = append(args, parsed)
		}
	}
	if opts.Mentionable {
		where = append(where, "INSTR(s.modewant,'J')>0 AND INSTR(s.modegiven,'J')>0")
	}
	args = append(args, opts.Limit+1)
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	rows, err := a.db.QueryxContext(ctx, tenantShardSQL(id, `SELECT STRAIGHT_JOIN gm.id,gm.topic,gm.user_id,gm.role,gm.member_status,gm.nickname,
			gm.pinned,gm.mute_notifications,gm.join_source,gm.joined_by,gm.joined_at,gm.left_at,gm.removed_by,
			gm.version,gm.created_at,gm.updated_at,s.modewant,s.modegiven,u.trusted,
			p.nickname AS profile_nickname,p.avatar_file_id AS profile_avatar_file_id,p.signature AS profile_signature,
			p.gender AS profile_gender,p.region AS profile_region,p.public_account AS profile_public_account
		FROM im_group_members AS gm FORCE INDEX (`+memberIndex+`)
		JOIN im_subscriptions AS s ON s.tenant_id=gm.tenant_id AND s.topic=gm.topic AND s.userid=gm.user_id AND s.deletedat IS NULL
		JOIN im_users AS u ON u.tenant_id=gm.tenant_id AND u.id=gm.user_id AND u.state!=?
		LEFT JOIN im_user_profiles AS p ON p.tenant_id=gm.tenant_id AND p.user_id=gm.user_id
		WHERE `+strings.Join(where, " AND ")+` ORDER BY gm.id LIMIT ?`), append([]any{t.StateDeleted}, args...)...)
	if err != nil {
		return nil, "", err
	}
	defer rows.Close()
	items := make([]t.GroupMember, 0, opts.Limit)
	next := ""
	for rows.Next() {
		row := struct {
			t.GroupMember
			Trusted              any
			ProfileNickname      sql.NullString `db:"profile_nickname"`
			ProfileAvatarFileID  sql.NullString `db:"profile_avatar_file_id"`
			ProfileSignature     sql.NullString `db:"profile_signature"`
			ProfileGender        sql.NullString `db:"profile_gender"`
			ProfileRegion        sql.NullString `db:"profile_region"`
			ProfilePublicAccount sql.NullString `db:"profile_public_account"`
		}{}
		if err = rows.StructScan(&row); err != nil {
			return nil, "", err
		}
		item := row.GroupMember
		scanGroupMember(&item)
		if len(items) == opts.Limit {
			if len(items) > 0 {
				next = strconv.FormatInt(items[len(items)-1].ID, 10)
			}
			continue
		}
		item.Public = profilePublicFromFields(row.ProfileNickname, row.ProfileAvatarFileID, row.ProfileSignature, row.ProfileGender, row.ProfileRegion, row.ProfilePublicAccount)
		item.Trusted = common.FromJSON(row.Trusted)
		items = append(items, item)
	}
	return items, next, rows.Err()
}

// TenantGroupMembersCount counts active members for the given topics using the
// same visibility rules as TenantGroupMemberList: membership active, subscription
// not deleted, user not deleted. Returns counts keyed by topic; topics with no
// active members are absent from the map.
func (a *adapter) TenantGroupMembersCount(id t.TenantID, topics []string) (map[string]int64, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return nil, err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupMembersCount(id, topics)
	}
	if len(topics) == 0 {
		return map[string]int64{}, nil
	}
	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(topics)), ",")
	args := make([]any, 0, len(topics)+3)
	args = append(args, t.StateDeleted, id, t.GroupMemberStatusActive)
	for _, topic := range topics {
		args = append(args, topic)
	}
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	rows, err := a.db.QueryContext(ctx, tenantShardSQL(id, `SELECT gm.topic, COUNT(DISTINCT gm.user_id) FROM im_group_members AS gm
		JOIN im_subscriptions AS s ON s.tenant_id=gm.tenant_id AND s.topic=gm.topic AND s.userid=gm.user_id AND s.deletedat IS NULL
		JOIN im_users AS u ON u.tenant_id=gm.tenant_id AND u.id=gm.user_id AND u.state!=?
		WHERE gm.tenant_id=? AND gm.member_status=? AND gm.topic IN (`+placeholders+`)
		GROUP BY gm.topic`), args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	counts := make(map[string]int64, len(topics))
	for rows.Next() {
		var topic string
		var count int64
		if err = rows.Scan(&topic, &count); err != nil {
			return nil, err
		}
		counts[topic] = count
	}
	return counts, rows.Err()
}

func (a *adapter) TenantUsersShareOpenPrivateGroup(id t.TenantID, userID, peerID t.Uid) (bool, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return false, err
	} else if tenantAdp != a {
		return tenantAdp.TenantUsersShareOpenPrivateGroup(id, userID, peerID)
	}
	if userID.IsZero() || peerID.IsZero() || userID == peerID {
		return false, t.ErrMalformed
	}
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	var exists int
	err := a.db.GetContext(ctx, &exists, tenantShardSQL(id, `
		SELECT 1
		FROM im_group_members AS gm1
		INNER JOIN im_group_members AS gm2
			ON gm2.tenant_id=gm1.tenant_id AND gm2.topic=gm1.topic
		INNER JOIN im_group_settings AS gs
			ON gs.tenant_id=gm1.tenant_id AND gs.topic=gm1.topic
		WHERE gm1.tenant_id=? AND gm1.user_id=? AND gm1.member_status=?
			AND gm2.user_id=? AND gm2.member_status=?
			AND gs.group_status=? AND gs.member_private_chat=1
		LIMIT 1`),
		id, decodeUID(userID), t.GroupMemberStatusActive,
		decodeUID(peerID), t.GroupMemberStatusActive, t.GroupStatusActive)
	if err == sql.ErrNoRows {
		return false, nil
	}
	return exists == 1, 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, id, topic, decodeUID(uid))
	where := "tenant_id=? AND topic=? AND user_id=?"
	if patch.Version > 0 {
		where += " AND version=?"
		args = append(args, patch.Version)
	}
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	res, err := tx.ExecContext(ctx, tenantShardSQL(id, "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
	}
	if err = recountGroupMembersTx(tx, id, topic); err != nil {
		return nil, err
	}
	if err = tx.Commit(); err != nil {
		return nil, err
	}
	return a.groupMemberByQuery(ctx, tenantShardSQL(id, "SELECT "+groupMemberSelectColumns+" FROM im_group_members WHERE tenant_id=? AND topic=? AND user_id=?"), 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.GroupInvitationStatusPending
	}
	if inv.ExpiresAt.IsZero() {
		inv.ExpiresAt = now.Add(7 * 24 * time.Hour)
	}
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	if strings.TrimSpace(inv.JoinRequestID) != "" {
		existing, err := a.groupInvitationByQuery(ctx, `SELECT inv.tenant_id,inv.id,inv.invitation_id,inv.topic,inv.inviter_user_id,inv.target_user_id,inv.message,inv.status,inv.result,
				inv.join_request_id,inv.idempotency_key,inv.expires_at,inv.handled_at,inv.created_at,inv.updated_at
			FROM im_group_invitations AS inv
			INNER JOIN im_group_join_requests AS req ON req.tenant_id=inv.tenant_id AND req.request_id=inv.join_request_id
			WHERE inv.tenant_id=? AND inv.topic=? AND inv.target_user_id=? AND inv.status=? AND req.status=?
			ORDER BY req.id DESC LIMIT 1`,
			id, inv.Topic, decodeUID(inv.TargetUserID), t.GroupInvitationStatusPending, t.GroupJoinRequestStatusPending)
		if err == nil {
			return existing, nil
		}
		if err != t.ErrNotFound {
			return nil, err
		}
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	_, err = tx.ExecContext(ctx, `INSERT INTO im_group_invitations(tenant_id,invitation_id,topic,inviter_user_id,target_user_id,message,status,result,join_request_id,idempotency_key,expires_at,handled_at,created_at,updated_at)
		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
		id, inv.InvitationID, inv.Topic, decodeUID(inv.InviterUserID), decodeUID(inv.TargetUserID), inv.Message, inv.Status, inv.Result, inv.JoinRequestID, inv.IdempotencyKey, inv.ExpiresAt, inv.HandledAt, now, now)
	if err != nil {
		return nil, err
	}
	if strings.TrimSpace(inv.JoinRequestID) != "" {
		_, err = tx.ExecContext(ctx, `INSERT INTO im_group_join_requests(tenant_id,request_id,topic,applicant_user_id,inviter_user_id,source,qr_token_hash,message,status,reject_reason,expires_at,idempotency_key,version,created_at,updated_at)
			VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
			id, inv.JoinRequestID, inv.Topic, decodeUID(inv.TargetUserID), nullableUID(inv.InviterUserID), "invitation", "", inv.Message, t.GroupJoinRequestStatusPending, "", inv.ExpiresAt, inv.IdempotencyKey, 1, now, now)
		if err != nil {
			return nil, err
		}
	}
	if err = tx.Commit(); err != nil {
		return nil, err
	}
	return inv, nil
}

func (a *adapter) TenantGroupInvitationAccept(id t.TenantID, invitationID string, uid t.Uid, memberPrivate any, now time.Time) (*t.GroupInvitation, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return nil, err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupInvitationAccept(id, invitationID, uid, memberPrivate, now)
	}
	if strings.TrimSpace(invitationID) == "" || uid.IsZero() || now.IsZero() {
		return nil, t.ErrMalformed
	}
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	var inv t.GroupInvitation
	var handledAt sql.NullTime
	err = tx.QueryRowxContext(ctx, `SELECT tenant_id,id,invitation_id,topic,inviter_user_id,target_user_id,message,status,result,
			join_request_id,idempotency_key,expires_at,handled_at,created_at,updated_at
		FROM im_group_invitations WHERE tenant_id=? AND invitation_id=? FOR UPDATE`, id, strings.TrimSpace(invitationID)).
		Scan(&inv.TenantID, &inv.ID, &inv.InvitationID, &inv.Topic, &inv.InviterUserRaw, &inv.TargetUserRaw, &inv.Message, &inv.Status,
			&inv.Result, &inv.JoinRequestID, &inv.IdempotencyKey, &inv.ExpiresAt, &handledAt, &inv.CreatedAt, &inv.UpdatedAt)
	if err == sql.ErrNoRows {
		return nil, t.ErrNotFound
	}
	if err != nil {
		return nil, err
	}
	inv.InviterUserID = store.EncodeUid(inv.InviterUserRaw)
	inv.TargetUserID = store.EncodeUid(inv.TargetUserRaw)
	if handledAt.Valid {
		inv.HandledAt = &handledAt.Time
	}
	if inv.TargetUserID != uid {
		return nil, t.ErrPermissionDenied
	}
	if inv.ExpiresAt.Before(now) {
		return nil, t.ErrExpired
	}
	if inv.Status == t.GroupInvitationStatusJoined || inv.Result == "joined" {
		return &inv, nil
	}
	if strings.TrimSpace(inv.JoinRequestID) != "" {
		if err = tx.Commit(); err != nil {
			return nil, err
		}
		return &inv, nil
	}
	var deletedAt sql.NullTime
	err = tx.QueryRowxContext(ctx, tenantShardSQL(id, "SELECT deletedat FROM im_subscriptions WHERE tenant_id=? AND topic=? AND userid=?"),
		id, inv.Topic, decodeUID(uid)).Scan(&deletedAt)
	if err != nil && err != sql.ErrNoRows {
		return nil, err
	}
	if err == sql.ErrNoRows || deletedAt.Valid {
		sub := &t.Subscription{TenantID: id, User: uid.String(), Topic: inv.Topic,
			ModeWant: t.ModeCPublic, ModeGiven: t.ModeCPublic, Private: memberPrivate}
		sub.InitTimes()
		if err = createTenantSubscription(tx, id, sub, deletedAt.Valid); err != nil {
			return nil, err
		}
	}
	member := &t.GroupMember{Topic: inv.Topic, UserID: uid, Role: t.GroupMemberRoleMember,
		MemberStatus: t.GroupMemberStatusActive, JoinSource: "invitation", JoinedBy: inv.InviterUserID, JoinedAt: now,
		Version: 1, CreatedAt: now, UpdatedAt: now}
	if err = upsertGroupMemberTx(tx, id, member); err != nil {
		return nil, err
	}
	if err = recountGroupMembersAndBumpVersionTx(tx, id, inv.Topic); err != nil {
		return nil, err
	}
	if err = updateGroupTopicSubCountTx(tx, id, inv.Topic); err != nil {
		return nil, err
	}
	if _, err = tx.ExecContext(ctx, `UPDATE im_group_invitations
		SET status=?,result=?,handled_at=?,updated_at=? WHERE tenant_id=? AND invitation_id=?`,
		t.GroupInvitationStatusJoined, "joined", now, now, id, inv.InvitationID); err != nil {
		return nil, err
	}
	if strings.TrimSpace(inv.JoinRequestID) != "" {
		if _, err = tx.ExecContext(ctx, `UPDATE im_group_join_requests
			SET status=?,handled_by=?,handled_at=?,updated_at=?,version=version+1
			WHERE tenant_id=? AND topic=? AND request_id=? AND status=?`,
			t.GroupJoinRequestStatusAccepted, decodeUID(uid), now, now, id, inv.Topic, inv.JoinRequestID, t.GroupJoinRequestStatusPending); err != nil {
			return nil, err
		}
	}
	if err = tx.Commit(); err != nil {
		return nil, err
	}
	inv.Status = t.GroupInvitationStatusJoined
	inv.Result = "joined"
	inv.HandledAt = &now
	inv.UpdatedAt = now
	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.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	existing, err := a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE tenant_id=? AND topic=? AND applicant_user_id=? AND status=? ORDER BY id DESC LIMIT 1",
		id, req.Topic, decodeUID(req.ApplicantUserID), t.GroupJoinRequestStatusPending)
	if err == nil {
		return existing, nil
	}
	if err != t.ErrNotFound {
		return nil, err
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	_, err = tx.ExecContext(ctx, `INSERT INTO im_group_join_requests(tenant_id,request_id,topic,applicant_user_id,inviter_user_id,source,qr_token_hash,message,status,reject_reason,expires_at,idempotency_key,version,created_at,updated_at)
		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
		id, req.RequestID, req.Topic, decodeUID(req.ApplicantUserID), nullableUID(req.InviterUserID), req.Source, req.QRTokenHash, req.Message, req.Status, req.RejectReason, req.ExpiresAt, req.IdempotencyKey, req.Version, now, now)
	if err != nil {
		return nil, err
	}
	if err = tx.Commit(); err != nil {
		return nil, err
	}
	return a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE tenant_id=? AND request_id=?", id, req.RequestID)
}

func (a *adapter) TenantGroupJoinRequestList(id t.TenantID, topic, status, cursor string, limit int) ([]t.GroupJoinRequest, string, error) {
	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{"r.tenant_id=?", "r.topic=?"}
	args := []any{id, topic}
	if status != "" && status != "all" {
		where = append(where, "r.status=?")
		args = append(args, status)
		if status == t.GroupJoinRequestStatusPending {
			where = append(where, `NOT EXISTS (
				SELECT 1 FROM im_group_join_requests AS newer
				WHERE newer.tenant_id=r.tenant_id AND newer.topic=r.topic AND newer.applicant_user_id=r.applicant_user_id
					AND newer.status=? AND newer.id>r.id
			)`)
			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, "r.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 r.* FROM im_group_join_requests AS r WHERE "+strings.Join(where, " AND ")+" ORDER BY r.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, memberPrivate any, 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, memberPrivate, now)
	}
	status := t.GroupJoinRequestStatusRejected
	if action == "accept" {
		status = t.GroupJoinRequestStatusAccepted
	}
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	var current t.GroupJoinRequest
	err = tx.QueryRowxContext(ctx, "SELECT * FROM im_group_join_requests WHERE tenant_id=? AND topic=? AND request_id=? FOR UPDATE", id, topic, requestID).StructScan(&current)
	if err == sql.ErrNoRows {
		return nil, t.ErrNotFound
	}
	if err != nil {
		return nil, err
	}
	scanGroupJoinRequest(&current)
	if current.Status != t.GroupJoinRequestStatusPending {
		return nil, t.ErrInvalidResponse
	}
	if version > 0 && current.Version != version {
		return nil, t.ErrInvalidResponse
	}
	if action == "accept" {
		sub := &t.Subscription{TenantID: id, User: current.ApplicantUserID.String(), Topic: topic,
			ModeWant: t.ModeCPublic, ModeGiven: t.ModeCPublic, Private: memberPrivate}
		sub.InitTimes()
		if err = createTenantSubscription(tx, id, sub, false); err != nil {
			return nil, err
		}
		member := &t.GroupMember{Topic: topic, UserID: current.ApplicantUserID, Role: t.GroupMemberRoleMember,
			MemberStatus: t.GroupMemberStatusActive, JoinSource: "approval", JoinedBy: actor, JoinedAt: now,
			Version: 1, CreatedAt: now, UpdatedAt: now}
		if err = upsertGroupMemberTx(tx, id, member); err != nil {
			return nil, err
		}
	}
	if _, err = tx.ExecContext(ctx, `UPDATE im_group_join_requests SET status=?,reject_reason=?,handled_by=?,handled_at=?,updated_at=?,version=version+1
		WHERE tenant_id=? AND topic=? AND request_id=? AND status=?`, status, rejectReason, decodeUID(actor), now, now, id, topic, requestID, t.GroupJoinRequestStatusPending); err != nil {
		return nil, err
	}
	if action == "accept" {
		if err = recountGroupMembersAndBumpVersionTx(tx, id, topic); err != nil {
			return nil, err
		}
		if err = updateGroupTopicSubCountTx(tx, id, topic); err != nil {
			return nil, err
		}
	}
	if err = tx.Commit(); err != nil {
		return nil, err
	}
	return a.groupJoinRequestByQuery(ctx, "SELECT * FROM im_group_join_requests WHERE tenant_id=? AND topic=? AND request_id=?", id, topic, requestID)
}

func (a *adapter) TenantGroupAnnouncementLatest(id t.TenantID, topic string) (*t.GroupAnnouncement, error) {
	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 "+groupAnnouncementSelectColumns(id)+" FROM im_group_announcements WHERE tenant_id=? AND topic=? AND status=? ORDER BY version DESC,id DESC LIMIT 1", id, topic, t.GroupAnnouncementStatusCurrent)
}

func (a *adapter) TenantGroupAnnouncementGet(id t.TenantID, topic, announcementID string) (*t.GroupAnnouncement, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return nil, err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupAnnouncementGet(id, topic, announcementID)
	}
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	return a.groupAnnouncementByQuery(ctx, "SELECT "+groupAnnouncementSelectColumns(id)+" FROM im_group_announcements WHERE tenant_id=? AND topic=? AND announcement_id=? LIMIT 1", id, topic, announcementID)
}

func (a *adapter) TenantGroupAnnouncementPut(id t.TenantID, ann *t.GroupAnnouncement, expectedVersion int) (*t.GroupAnnouncement, error) {
	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 tenant_id=? AND topic=?", id, 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 tenant_id=? AND topic=? AND status=?", t.GroupAnnouncementStatusSuperseded, now, id, 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(tenant_id,announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at)
		VALUES(?,?,?,?,?,?,?,?,?,?,?)`, id, 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()
	}
	readID, err := nextSnowflakeID()
	if err != nil {
		return err
	}
	_, err = a.db.ExecContext(ctx, `INSERT INTO im_group_announcement_reads(id,tenant_id,announcement_id,topic,user_id,read_at)
		VALUES(?,?,?,?,?,?) ON DUPLICATE KEY UPDATE read_at=VALUES(read_at),updated_at=CURRENT_TIMESTAMP(3)`, readID, id, 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{"tenant_id=?", "topic=?", "status=?"}
	args := []any{id, 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 "+groupMuteSelectColumns+" 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) TenantGroupMuteGet(id t.TenantID, topic string, uid t.Uid) (*t.GroupMute, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return nil, err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupMuteGet(id, topic, uid)
	}
	if topic == "" || uid.IsZero() {
		return nil, t.ErrMalformed
	}
	now := time.Now().UTC()
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	return a.groupMuteByQuery(ctx, "SELECT "+groupMuteSelectColumns+" FROM im_group_mutes WHERE tenant_id=? AND topic=? AND user_id=? AND status=? AND (expires_at IS NULL OR expires_at>?)",
		id, topic, decodeUID(uid), t.GroupMuteStatusActive, now)
}

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.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	muteID, err := nextSnowflakeID()
	if err != nil {
		return nil, err
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO im_group_mutes(id,tenant_id,topic,user_id,muted_by,muted_at,expires_at,reason,status,version)
		VALUES(?,?,?,?,?,?,?,?,?,1)
		ON DUPLICATE KEY UPDATE muted_by=VALUES(muted_by),muted_at=VALUES(muted_at),expires_at=VALUES(expires_at),reason=VALUES(reason),status=VALUES(status),revoked_by=NULL,revoked_at=NULL,version=version+1,updated_at=CURRENT_TIMESTAMP(3)`,
		muteID, id, mute.Topic, decodeUID(mute.UserID), decodeUID(mute.MutedBy), now, mute.ExpiresAt, mute.Reason, mute.Status)
	if err != nil {
		return nil, err
	}
	if err = tx.Commit(); err != nil {
		return nil, err
	}
	return a.groupMuteByQuery(ctx, "SELECT "+groupMuteSelectColumns+" FROM im_group_mutes WHERE tenant_id=? AND topic=? AND user_id=?", id, mute.Topic, decodeUID(mute.UserID))
}

func (a *adapter) TenantGroupMuteDelete(id t.TenantID, topic string, uid, actor t.Uid, now time.Time) error {
	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.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	res, err := tx.ExecContext(ctx, "UPDATE im_group_mutes SET status=?,revoked_by=?,revoked_at=?,version=version+1,updated_at=? WHERE tenant_id=? AND topic=? AND user_id=? AND status=?",
		t.GroupMuteStatusRevoked, decodeUID(actor), now, now, id, topic, decodeUID(uid), t.GroupMuteStatusActive)
	if err != nil {
		return err
	}
	if affected, _ := res.RowsAffected(); affected == 0 {
		return t.ErrNotFound
	}
	return tx.Commit()
}

func (a *adapter) TenantGroupFileList(id t.TenantID, topic, fileType, query, cursor string, limit int) ([]t.GroupFile, string, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return nil, "", err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupFileList(id, topic, fileType, query, cursor, limit)
	}
	if strings.TrimSpace(topic) == "" {
		return nil, "", t.ErrMalformed
	}
	where := []string{"tenant_id=?", "topic=?", "status='active'"}
	args := []any{id, strings.TrimSpace(topic)}
	if fileType != "" {
		where = append(where, "file_type=?")
		args = append(args, strings.TrimSpace(fileType))
	}
	if query != "" {
		where = append(where, "file_name LIKE ?")
		args = append(args, "%"+strings.TrimSpace(query)+"%")
	}
	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 tenant_id,id,group_file_id,topic,file_id,source_topic,source_seq,uploader_id,
			file_name,mime_type,file_type,file_size,status,deleted_by,deleted_at,idempotency_key,created_at,updated_at
		FROM im_group_files WHERE `+strings.Join(where, " AND ")+` ORDER BY id DESC LIMIT ?`, args...)
	if err != nil {
		return nil, "", err
	}
	defer rows.Close()
	items := make([]t.GroupFile, 0, limit)
	next := ""
	for rows.Next() {
		file, err := scanGroupFile(rows)
		if err != nil {
			return nil, "", err
		}
		if len(items) == limit {
			next = strconv.FormatInt(file.ID, 10)
			continue
		}
		a.enrichGroupFileFromUpload(id, file)
		items = append(items, *file)
	}
	return items, next, rows.Err()
}

func (a *adapter) TenantGroupFileCreate(id t.TenantID, file *t.GroupFile) (*t.GroupFile, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return nil, err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupFileCreate(id, file)
	}
	if file == nil || strings.TrimSpace(file.Topic) == "" || strings.TrimSpace(file.FileID) == "" || file.UploaderID.IsZero() {
		return nil, t.ErrMalformed
	}
	if file.GroupFileID == "" {
		file.GroupFileID = "gfile_" + store.Store.GetUidString()
	}
	if file.FileType == "" {
		file.FileType = "file"
	}
	if file.Status == "" {
		file.Status = "active"
	}
	if file.IdempotencyKey == "" {
		file.IdempotencyKey = file.GroupFileID
	}
	a.enrichGroupFileFromUpload(id, file)
	now := t.TimeNow()
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	_, err = tx.ExecContext(ctx, `INSERT INTO im_group_files
		(tenant_id,group_file_id,topic,file_id,source_topic,source_seq,uploader_id,file_name,mime_type,file_type,file_size,status,idempotency_key,created_at,updated_at)
		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
		ON DUPLICATE KEY UPDATE
			file_name=IF(file_name='', VALUES(file_name), file_name),
			mime_type=IF(mime_type='', VALUES(mime_type), mime_type),
			file_size=IF(file_size=0, VALUES(file_size), file_size),
			file_type=IF(file_type='' OR file_type='file', VALUES(file_type), file_type),
			updated_at=updated_at`,
		id, file.GroupFileID, file.Topic, file.FileID, file.SourceTopic, file.SourceSeq, decodeUID(file.UploaderID),
		file.FileName, file.MIMEType, file.FileType, file.FileSize, file.Status, file.IdempotencyKey, now, now)
	if err != nil && !isDupe(err) {
		return nil, err
	}
	uploadID := strings.TrimSpace(file.FileID)
	if dot := strings.IndexByte(uploadID, '.'); dot > 0 {
		uploadID = uploadID[:dot]
	}
	if fid := t.ParseUid(uploadID); !fid.IsZero() {
		if a.tenantScopedDB {
			_, err = tx.ExecContext(ctx, `INSERT INTO im_filemsglinks(tenant_id,createdat,fileid,topic)
				SELECT ?,?,?,? WHERE NOT EXISTS (
					SELECT 1 FROM im_filemsglinks WHERE tenant_id=? AND fileid=? AND topic=?)`,
				id, now, decodeUID(fid), file.Topic, id, decodeUID(fid), file.Topic)
		} else {
			_, err = tx.ExecContext(ctx, `INSERT INTO im_filemsglinks(tenant_id,createdat,fileid,topic)
				SELECT ?,?,?,? WHERE NOT EXISTS (
					SELECT 1 FROM im_filemsglinks WHERE tenant_id=? AND fileid=? AND topic=?)`,
				id, now, decodeUID(fid), file.Topic, id, decodeUID(fid), file.Topic)
		}
		if err != nil {
			return nil, err
		}
	}
	if err = tx.Commit(); err != nil {
		return nil, err
	}
	created, err := a.groupFileByQuery(ctx, "SELECT * FROM im_group_files WHERE tenant_id=? AND topic=? AND file_id=?", id, file.Topic, file.FileID)
	if err != nil {
		return nil, err
	}
	a.enrichGroupFileFromUpload(id, created)
	return created, nil
}

func (a *adapter) enrichGroupFileFromUpload(id t.TenantID, file *t.GroupFile) {
	if file == nil {
		return
	}
	fileID := strings.TrimSpace(file.FileID)
	if dot := strings.IndexByte(fileID, '.'); dot > 0 {
		fileID = fileID[:dot]
	}
	if t.ParseUid(fileID).IsZero() {
		return
	}
	fd, err := a.TenantFileGet(id, fileID)
	if err != nil || fd == nil {
		if file.FileName == "" {
			file.FileName = file.FileID
		}
		return
	}
	if file.MIMEType == "" {
		file.MIMEType = strings.TrimSpace(fd.MimeType)
	}
	if file.FileSize <= 0 {
		file.FileSize = fd.Size
	}
	if file.FileName == "" {
		if original := strings.TrimSpace(fd.OriginalName); original != "" {
			file.FileName = original
		} else {
			file.FileName = file.FileID
		}
	}
}

func (a *adapter) TenantGroupFileDelete(id t.TenantID, topic, groupFileID string, actor t.Uid, canManage bool, now time.Time) (*t.GroupFile, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return nil, err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupFileDelete(id, topic, groupFileID, actor, canManage, now)
	}
	if strings.TrimSpace(topic) == "" || strings.TrimSpace(groupFileID) == "" || actor.IsZero() || now.IsZero() {
		return nil, t.ErrMalformed
	}
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	query := `UPDATE im_group_files
		SET status='deleted',deleted_by=?,deleted_at=?,updated_at=?
		WHERE tenant_id=? AND topic=? AND group_file_id=? AND status='active'`
	args := []any{decodeUID(actor), now, now, id, strings.TrimSpace(topic), strings.TrimSpace(groupFileID)}
	if !canManage {
		query += " AND uploader_id=?"
		args = append(args, decodeUID(actor))
	}
	res, err := a.db.ExecContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	if affected, _ := res.RowsAffected(); affected == 0 {
		return nil, t.ErrNotFound
	}
	return a.groupFileByQuery(ctx, "SELECT * FROM im_group_files WHERE tenant_id=? AND topic=? AND group_file_id=?", id, strings.TrimSpace(topic), strings.TrimSpace(groupFileID))
}

func (a *adapter) TenantGroupCreate(id t.TenantID, topic *t.Topic, owner t.Uid, ownerPrivate any,
	settings *t.GroupSettings, members []t.Uid, memberPrivate any, now time.Time) error {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupCreate(id, topic, owner, ownerPrivate, settings, members, memberPrivate, now)
	}
	if topic == nil || settings == nil || topic.Id == "" || settings.Topic != topic.Id || owner.IsZero() {
		return t.ErrMalformed
	}
	topic.TenantID = id
	topic.Owner = owner.String()
	settings.TenantID = id
	settings.OwnerUserID = owner
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if err = tenantTopicCreate(tx, topic); err != nil {
		return err
	}
	ownerSub := &t.Subscription{TenantID: id, ObjHeader: t.ObjHeader{CreatedAt: topic.CreatedAt}, User: owner.String(),
		Topic: topic.Id, ModeGiven: t.ModeCFull, ModeWant: topic.GetAccess(owner), Private: ownerPrivate}
	ownerSub.InitTimes()
	if err = createTenantSubscription(tx, id, ownerSub, false); err != nil {
		return err
	}
	settings.MemberCount = 1
	if err = insertGroupSettingsTx(tx, id, settings); err != nil {
		return err
	}
	ownerMember := &t.GroupMember{Topic: topic.Id, UserID: owner, Role: t.GroupMemberRoleOwner,
		MemberStatus: t.GroupMemberStatusActive, JoinSource: "create", JoinedBy: owner, JoinedAt: now,
		Version: 1, CreatedAt: now, UpdatedAt: now}
	if err = upsertGroupMemberTx(tx, id, ownerMember); err != nil {
		return err
	}
	seen := map[t.Uid]bool{owner: true}
	for _, uid := range members {
		if uid.IsZero() || seen[uid] {
			continue
		}
		seen[uid] = true
		sub := &t.Subscription{TenantID: id, User: uid.String(), Topic: topic.Id,
			ModeWant: t.ModeCPublic, ModeGiven: t.ModeCPublic, Private: memberPrivate}
		sub.InitTimes()
		if err = createTenantSubscription(tx, id, sub, false); err != nil {
			return err
		}
		member := &t.GroupMember{Topic: topic.Id, UserID: uid, Role: t.GroupMemberRoleMember,
			MemberStatus: t.GroupMemberStatusActive, JoinSource: "create", JoinedBy: owner, JoinedAt: now,
			Version: 1, CreatedAt: now, UpdatedAt: now}
		if err = upsertGroupMemberTx(tx, id, member); err != nil {
			return err
		}
	}
	if err = recountGroupMembersTx(tx, id, topic.Id); err != nil {
		return err
	}
	if err = updateGroupTopicSubCountTx(tx, id, topic.Id); err != nil {
		return err
	}
	return tx.Commit()
}

func (a *adapter) TenantGroupInviteMember(id t.TenantID, topic string, inviter, target t.Uid,
	message, idempotencyKey string, private any, now time.Time) (bool, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return false, err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupInviteMember(id, topic, inviter, target, message, idempotencyKey, private, now)
	}
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return false, err
	}
	defer tx.Rollback()
	var deletedAt sql.NullTime
	err = tx.QueryRowContext(ctx, tenantShardSQL(id, "SELECT deletedat FROM im_subscriptions WHERE tenant_id=? AND topic=? AND userid=? FOR UPDATE"), id, topic, decodeUID(target)).Scan(&deletedAt)
	if err != nil && err != sql.ErrNoRows {
		return false, err
	}
	if err == nil && !deletedAt.Valid {
		return false, nil
	}
	sub := &t.Subscription{TenantID: id, User: target.String(), Topic: topic,
		ModeWant: t.ModeCPublic, ModeGiven: t.ModeCPublic, Private: private}
	sub.InitTimes()
	if err = createTenantSubscription(tx, id, sub, false); err != nil {
		return false, err
	}
	invitationID := "ginv_" + store.Store.GetUidString()
	if idempotencyKey == "" {
		idempotencyKey = invitationID
	} else {
		idempotencyKey += ":" + target.UserId()
	}
	if _, err = tx.ExecContext(ctx, `INSERT INTO im_group_invitations
		(tenant_id,invitation_id,topic,inviter_user_id,target_user_id,message,status,result,idempotency_key,expires_at,handled_at,created_at,updated_at)
		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, id, invitationID, topic, decodeUID(inviter), decodeUID(target), message,
		t.GroupInvitationStatusJoined, "joined", idempotencyKey, now.Add(7*24*time.Hour), now, now, now); err != nil {
		return false, err
	}
	member := &t.GroupMember{Topic: topic, UserID: target, Role: t.GroupMemberRoleMember,
		MemberStatus: t.GroupMemberStatusActive, JoinSource: "invitation", JoinedBy: inviter, JoinedAt: now,
		Version: 1, CreatedAt: now, UpdatedAt: now}
	if err = upsertGroupMemberTx(tx, id, member); err != nil {
		return false, err
	}
	if err = recountGroupMembersAndBumpVersionTx(tx, id, topic); err != nil {
		return false, err
	}
	if err = updateGroupTopicSubCountTx(tx, id, topic); err != nil {
		return false, err
	}
	if err = tx.Commit(); err != nil {
		return false, err
	}
	return true, nil
}

func (a *adapter) TenantGroupLeave(id t.TenantID, topic string, uid t.Uid, now time.Time) error {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupLeave(id, topic, uid, now)
	}
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	res, err := tx.ExecContext(ctx, tenantShardSQL(id, "UPDATE im_subscriptions SET updatedat=?,deletedat=? WHERE tenant_id=? AND topic=? AND userid=? AND deletedat IS NULL"), now, now, id, topic, decodeUID(uid))
	if err != nil {
		return err
	}
	if affected, _ := res.RowsAffected(); affected == 0 {
		return t.ErrNotFound
	}
	if _, err = tx.ExecContext(ctx, tenantShardSQL(id, `UPDATE im_group_members SET member_status=?,left_at=?,version=version+1,updated_at=?
		WHERE tenant_id=? AND topic=? AND user_id=? AND member_status=?`), t.GroupMemberStatusLeft, now, now, id, topic, decodeUID(uid), t.GroupMemberStatusActive); err != nil {
		return err
	}
	if err = recountGroupMembersAndBumpVersionTx(tx, id, topic); err != nil {
		return err
	}
	if err = updateGroupTopicSubCountTx(tx, id, topic); err != nil {
		return err
	}
	return tx.Commit()
}

func (a *adapter) TenantGroupRoleUpdate(id t.TenantID, topic string, actor, target t.Uid,
	role string, modeGiven t.AccessMode, now time.Time) error {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupRoleUpdate(id, topic, actor, target, role, modeGiven, now)
	}
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	res, err := tx.ExecContext(ctx, tenantShardSQL(id, "UPDATE im_subscriptions SET modegiven=?,modewant=?,updatedat=? WHERE tenant_id=? AND topic=? AND userid=? AND deletedat IS NULL"),
		modeGiven.String(), modeGiven.String(), now, id, topic, decodeUID(target))
	if err != nil {
		return err
	}
	if affected, _ := res.RowsAffected(); affected == 0 {
		return t.ErrNotFound
	}
	res, err = tx.ExecContext(ctx, tenantShardSQL(id, "UPDATE im_group_members SET role=?,version=version+1,updated_at=? WHERE tenant_id=? AND topic=? AND user_id=? AND member_status=?"),
		role, now, id, topic, decodeUID(target), t.GroupMemberStatusActive)
	if err != nil {
		return err
	}
	if affected, _ := res.RowsAffected(); affected == 0 {
		return t.ErrNotFound
	}
	return tx.Commit()
}

func groupAdminAccessMode() t.AccessMode {
	return t.ModeCPublic | t.ModeApprove
}

func (a *adapter) TenantGroupOwnerTransfer(id t.TenantID, topic string, previousOwner, newOwner t.Uid,
	version int, now time.Time) error {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupOwnerTransfer(id, topic, previousOwner, newOwner, version, now)
	}
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTxx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	updates := []struct {
		uid  t.Uid
		mode t.AccessMode
		role string
	}{{previousOwner, groupAdminAccessMode(), t.GroupMemberRoleAdmin}, {newOwner, t.ModeCFull, t.GroupMemberRoleOwner}}
	for _, update := range updates {
		res, execErr := tx.ExecContext(ctx, tenantShardSQL(id, "UPDATE im_subscriptions SET modegiven=?,modewant=?,updatedat=? WHERE tenant_id=? AND topic=? AND userid=? AND deletedat IS NULL"),
			update.mode.String(), update.mode.String(), now, id, topic, decodeUID(update.uid))
		if execErr != nil {
			return execErr
		}
		if affected, _ := res.RowsAffected(); affected == 0 {
			return t.ErrNotFound
		}
		res, execErr = tx.ExecContext(ctx, tenantShardSQL(id, "UPDATE im_group_members SET role=?,version=version+1,updated_at=? WHERE tenant_id=? AND topic=? AND user_id=? AND member_status=?"),
			update.role, now, id, topic, decodeUID(update.uid), t.GroupMemberStatusActive)
		if execErr != nil {
			return execErr
		}
		if affected, _ := res.RowsAffected(); affected == 0 {
			return t.ErrNotFound
		}
	}
	if _, err = tx.ExecContext(ctx, tenantShardSQL(id, "UPDATE im_topics SET owner=?,updatedat=? WHERE tenant_id=? AND name=?"), decodeUID(newOwner), now, id, topic); err != nil {
		return err
	}
	args := []any{decodeUID(newOwner), now, id, topic}
	query := "UPDATE im_group_settings SET owner_user_id=?,version=version+1,updated_at=? WHERE tenant_id=? AND topic=?"
	if version > 0 {
		query += " AND version=?"
		args = append(args, version)
	}
	res, err := tx.ExecContext(ctx, query, args...)
	if err != nil {
		return err
	}
	if affected, _ := res.RowsAffected(); affected == 0 {
		return t.ErrInvalidResponse
	}
	return tx.Commit()
}

func (a *adapter) TenantGroupCanPost(id t.TenantID, topic string, uid t.Uid, admin bool, now time.Time) (bool, error) {
	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
		return false, err
	} else if tenantAdp != a {
		return tenantAdp.TenantGroupCanPost(id, topic, uid, admin, now)
	}
	if admin {
		return true, nil
	}
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	var muteAll, memberMuted bool
	err := a.db.QueryRowContext(ctx, `SELECT
		COALESCE((SELECT mute_all FROM im_group_settings WHERE tenant_id=? AND topic=? AND group_status=? LIMIT 1),0),
		EXISTS(SELECT 1 FROM im_group_mutes WHERE tenant_id=? AND topic=? AND user_id=? AND status=? AND (expires_at IS NULL OR expires_at>?))`,
		id, topic, t.GroupStatusActive, id, topic, decodeUID(uid), t.GroupMuteStatusActive, now).Scan(&muteAll, &memberMuted)
	if err != nil {
		return false, err
	}
	return !muteAll && !memberMuted, nil
}

func insertGroupSettingsTx(tx *sqlx.Tx, tenantID t.TenantID, settings *t.GroupSettings) error {
	settingsID, err := nextSnowflakeID()
	if err != nil {
		return err
	}
	_, err = tx.Exec(`INSERT INTO im_group_settings
		(id,tenant_id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,created_by,created_at,updated_at)
		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, settingsID, tenantID, 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)
	return err
}

func upsertGroupMemberTx(tx *sqlx.Tx, tenantID t.TenantID, member *t.GroupMember) error {
	_, err := tx.Exec(tenantShardSQL(tenantID, `INSERT INTO im_group_members
		(tenant_id,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=VALUES(joined_at),left_at=NULL,removed_by=NULL,version=version+1,updated_at=VALUES(updated_at)`),
		tenantID, 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)
	return err
}

func recountGroupMembersTx(tx *sqlx.Tx, tenantID t.TenantID, topic string) error {
	_, err := tx.Exec(tenantShardSQL(tenantID, `UPDATE im_group_settings SET member_count=(SELECT COUNT(*) FROM im_group_members
		WHERE tenant_id=? AND topic=? AND member_status=?),updated_at=CURRENT_TIMESTAMP(3) WHERE tenant_id=? AND topic=?`), tenantID, topic, t.GroupMemberStatusActive, tenantID, topic)
	return err
}

func recountGroupMembersAndBumpVersionTx(tx *sqlx.Tx, tenantID t.TenantID, topic string) error {
	_, err := tx.Exec(tenantShardSQL(tenantID, `UPDATE im_group_settings SET member_count=(SELECT COUNT(*) FROM im_group_members
		WHERE tenant_id=? AND topic=? AND member_status=?),version=version+1,updated_at=CURRENT_TIMESTAMP(3) WHERE tenant_id=? AND topic=?`), tenantID, topic, t.GroupMemberStatusActive, tenantID, topic)
	return err
}

func updateGroupTopicSubCountTx(tx *sqlx.Tx, tenantID t.TenantID, topic string) error {
	_, err := tx.Exec(tenantShardSQL(tenantID, `UPDATE im_topics SET subcnt=(SELECT COUNT(*) FROM im_subscriptions
		WHERE tenant_id=? AND topic IN (?,?) AND deletedat IS NULL) WHERE tenant_id=? AND name=?`), tenantID, topic, t.GrpToChn(topic), tenantID, topic)
	return err
}

func syncGroupMuteAllTx(tx *sqlx.Tx, tenantID t.TenantID, topic string, muteAll bool, now time.Time) error {
	mode := t.ModeCPublic
	if muteAll {
		mode = t.ModeCPublic &^ t.ModeWrite
	}
	_, err := tx.Exec(tenantShardSQL(tenantID, `UPDATE im_subscriptions AS s
		JOIN im_group_members AS gm ON gm.tenant_id=s.tenant_id AND gm.topic=s.topic AND gm.user_id=s.userid
		SET s.modewant=?,s.modegiven=?,s.updatedat=?,
			gm.version=gm.version+1,gm.updated_at=?
		WHERE s.topic=? AND s.deletedat IS NULL
			AND gm.member_status=? AND gm.role=?`),
		mode.String(), mode.String(), now, now, topic, t.GroupMemberStatusActive, t.GroupMemberRoleMember)
	return err
}

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, tenantID t.TenantID, topic string) (*t.GroupSettings, error) {
	var item t.GroupSettings
	err := a.db.QueryRowxContext(ctx, "SELECT "+groupSettingsSelectColumns+" FROM im_group_settings WHERE tenant_id=? AND topic=?", tenantID, 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) groupInvitationByQuery(ctx context.Context, query string, args ...any) (*t.GroupInvitation, error) {
	var item t.GroupInvitation
	var handledAt sql.NullTime
	err := a.db.QueryRowxContext(ctx, query, args...).Scan(&item.TenantID, &item.ID, &item.InvitationID, &item.Topic, &item.InviterUserRaw, &item.TargetUserRaw,
		&item.Message, &item.Status, &item.Result, &item.JoinRequestID, &item.IdempotencyKey, &item.ExpiresAt, &handledAt, &item.CreatedAt, &item.UpdatedAt)
	if err == sql.ErrNoRows {
		return nil, t.ErrNotFound
	}
	if err != nil {
		return nil, err
	}
	scanGroupInvitation(&item, handledAt)
	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) groupFileByQuery(ctx context.Context, query string, args ...any) (*t.GroupFile, error) {
	rows, err := a.db.QueryxContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	if rows.Next() {
		return scanGroupFile(rows)
	}
	if err = rows.Err(); err != nil {
		return nil, err
	}
	return nil, t.ErrNotFound
}

func (a *adapter) recountGroupMembers(ctx context.Context, tenantID t.TenantID, topic string) error {
	_, err := a.db.ExecContext(ctx, tenantShardSQL(tenantID, "UPDATE im_group_settings SET member_count=(SELECT COUNT(*) FROM im_group_members WHERE tenant_id=? AND topic=? AND member_status=?),updated_at=CURRENT_TIMESTAMP(3) WHERE tenant_id=? AND topic=?"), tenantID, topic, t.GroupMemberStatusActive, tenantID, 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 scanGroupInvitation(item *t.GroupInvitation, handledAt sql.NullTime) {
	item.InviterUserID = store.EncodeUid(item.InviterUserRaw)
	item.TargetUserID = store.EncodeUid(item.TargetUserRaw)
	if handledAt.Valid {
		item.HandledAt = &handledAt.Time
	}
}

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)
	}
}

func scanGroupFile(rows *sqlx.Rows) (*t.GroupFile, error) {
	var item t.GroupFile
	if err := rows.StructScan(&item); err != nil {
		return nil, err
	}
	item.UploaderID = store.EncodeUid(item.UploaderRaw)
	if item.DeletedByRaw != nil {
		item.DeletedBy = store.EncodeUid(*item.DeletedByRaw)
	}
	return &item, nil
}