//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 = `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` groupAnnouncementSelectColumns = `id,announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at` groupMuteSelectColumns = `topic,user_id,muted_by,muted_at,expires_at,reason,status,revoked_by,revoked_at,version,created_at,updated_at` ) func (a *adapter) TenantGroupSettingsCreate(id t.TenantID, settings *t.GroupSettings) (*t.GroupSettings, error) { if tenantAdp, err := a.routeGroupTenant(id); err != nil { return nil, err } 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,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, 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.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, topic) where := "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, 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, 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 topic=? AND group_status<>'dissolved'`, decodeUID(actor), now, now, strings.TrimSpace(topic)) if err != nil { return nil, err } if affected, _ := res.RowsAffected(); affected == 0 { return nil, t.ErrNotFound } if _, err = tx.ExecContext(ctx, `UPDATE im_subscriptions SET updatedat=?,deletedat=? WHERE topic IN (?,?) AND deletedat IS NULL`, now, now, strings.TrimSpace(topic), t.GrpToChn(strings.TrimSpace(topic))); err != nil { return nil, err } if _, err = tx.ExecContext(ctx, `UPDATE im_group_members SET member_status=?,left_at=?,removed_by=?,version=version+1,updated_at=? WHERE topic=? AND member_status=?`, t.GroupMemberStatusLeft, now, decodeUID(actor), now, strings.TrimSpace(topic), t.GroupMemberStatusActive); err != nil { return nil, err } if _, err = tx.ExecContext(ctx, `UPDATE im_topics SET updatedat=?,touchedat=?,state=?,stateat=?,subcnt=0 WHERE name=?`, now, now, t.StateDeleted, now, strings.TrimSpace(topic)); err != nil { return nil, err } if err = tx.Commit(); err != nil { return nil, err } 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 } memberIndex := "idx_im_group_members_topic_status" where := []string{"gm.topic=?", "gm.member_status=?"} args := []any{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 JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.fn')) LIKE ? OR JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.name')) LIKE ? OR JSON_UNQUOTE(JSON_EXTRACT(u.public,'$.display_name')) LIKE ? OR u.username LIKE ?)`) args = append(args, like, like, like, 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, `SELECT STRAIGHT_JOIN gm.*,s.modewant,s.modegiven,u.public,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.topic=gm.topic AND s.userid=gm.user_id AND s.deletedat IS NULL JOIN im_users AS u ON u.id=gm.user_id AND u.state!=? LEFT JOIN im_user_profiles AS p ON 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 Public any 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(common.FromJSON(row.Public), 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() } 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.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_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, topic); err != nil { return nil, err } if err = tx.Commit(); err != nil { return nil, err } 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.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.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.request_id=inv.join_request_id WHERE inv.topic=? AND inv.target_user_id=? AND inv.status=? AND req.status=? ORDER BY req.id DESC LIMIT 1`, 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(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 } if strings.TrimSpace(inv.JoinRequestID) != "" { _, err = tx.ExecContext(ctx, `INSERT INTO im_group_join_requests(request_id,topic,applicant_user_id,inviter_user_id,source,qr_token_hash,message,status,reject_reason,expires_at,idempotency_key,version,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, 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 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 invitation_id=? FOR UPDATE`, strings.TrimSpace(invitationID)). Scan(&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, "SELECT deletedat FROM im_subscriptions WHERE topic=? AND userid=?", 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 = createSubscription(tx, 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, member); err != nil { return nil, err } if err = recountGroupMembersAndBumpVersionTx(tx, inv.Topic); err != nil { return nil, err } if err = updateGroupTopicSubCountTx(tx, inv.Topic); err != nil { return nil, err } if _, err = tx.ExecContext(ctx, `UPDATE im_group_invitations SET status=?,result=?,handled_at=?,updated_at=? WHERE invitation_id=?`, t.GroupInvitationStatusJoined, "joined", now, now, 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 topic=? AND request_id=? AND status=?`, t.GroupJoinRequestStatusAccepted, decodeUID(uid), now, now, 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 topic=? AND applicant_user_id=? AND status=? ORDER BY id DESC LIMIT 1", 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(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 } if err = tx.Commit(); 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{"r.topic=?"} args := []any{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.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 topic=? AND request_id=? FOR UPDATE", topic, requestID).StructScan(¤t) if err == sql.ErrNoRows { return nil, t.ErrNotFound } if err != nil { return nil, err } scanGroupJoinRequest(¤t) 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 = createSubscription(tx, 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, 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 topic=? AND request_id=? AND status=?`, status, rejectReason, decodeUID(actor), now, now, topic, requestID, t.GroupJoinRequestStatusPending); err != nil { return nil, err } if action == "accept" { if err = recountGroupMembersAndBumpVersionTx(tx, topic); err != nil { return nil, err } if err = updateGroupTopicSubCountTx(tx, 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 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 "+groupAnnouncementSelectColumns+" FROM im_group_announcements WHERE topic=? AND status=? ORDER BY version DESC,id DESC LIMIT 1", 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+" FROM im_group_announcements WHERE topic=? AND announcement_id=? LIMIT 1", 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 topic=?", ann.Topic).Scan(¤tVersion) if err != nil { return nil, err } if expectedVersion > 0 && currentVersion > 0 && expectedVersion != currentVersion { return nil, t.ErrInvalidResponse } version := currentVersion + 1 now := ann.PublishedAt if now.IsZero() { now = time.Now().UTC().Round(time.Millisecond) } if _, err = tx.ExecContext(ctx, "UPDATE im_group_announcements SET status=?,updated_at=? WHERE topic=? AND status=?", t.GroupAnnouncementStatusSuperseded, now, ann.Topic, t.GroupAnnouncementStatusCurrent); err != nil { return nil, err } if ann.AnnouncementID == "" { ann.AnnouncementID = "ann_" + store.Store.GetUidString() } if ann.IdempotencyKey == "" { ann.IdempotencyKey = ann.AnnouncementID } if _, err = tx.ExecContext(ctx, `INSERT INTO im_group_announcements(announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?)`, ann.AnnouncementID, ann.Topic, decodeUID(ann.AuthorUserID), ann.Content, t.GroupAnnouncementStatusCurrent, version, now, ann.IdempotencyKey, now, now); err != nil { return nil, err } if err = tx.Commit(); err != nil { return nil, err } return a.TenantGroupAnnouncementLatest(id, ann.Topic) } func (a *adapter) TenantGroupAnnouncementRead(id t.TenantID, topic, announcementID string, uid t.Uid, readAt time.Time) error { if tenantAdp, err := a.routeGroupTenant(id); err != nil { return err } else if tenantAdp != a { return tenantAdp.TenantGroupAnnouncementRead(id, topic, announcementID, uid, readAt) } ctx, cancel := a.getContext() if cancel != nil { defer cancel() } readID, err := nextSnowflakeID() if err != nil { return err } _, err = a.db.ExecContext(ctx, `INSERT INTO im_group_announcement_reads(id,announcement_id,topic,user_id,read_at) VALUES(?,?,?,?,?) ON DUPLICATE KEY UPDATE read_at=VALUES(read_at),updated_at=CURRENT_TIMESTAMP(3)`, readID, announcementID, topic, decodeUID(uid), readAt) return err } 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 "+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) 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,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, 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 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.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 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 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{"topic=?", "status='active'"} args := []any{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 0 { uploadID = uploadID[:dot] } if fid := t.ParseUid(uploadID); !fid.IsZero() { if a.tenantScopedDB { _, err = tx.ExecContext(ctx, `INSERT INTO im_filemsglinks(createdat,fileid,topic) SELECT ?,?,? WHERE NOT EXISTS ( SELECT 1 FROM im_filemsglinks WHERE fileid=? AND topic=?)`, now, decodeUID(fid), file.Topic, decodeUID(fid), file.Topic) } else { _, err = tx.ExecContext(ctx, `INSERT INTO im_filemsglinks(tenant_id,createdat,fileid,topic) 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 topic=? AND file_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 topic=? AND group_file_id=? AND status='active'` args := []any{decodeUID(actor), now, now, 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 topic=? AND group_file_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 } 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 = a.topicCreate(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 = createSubscription(tx, ownerSub, false); err != nil { return err } settings.MemberCount = 1 if err = insertGroupSettingsTx(tx, 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, 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 = createSubscription(tx, 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, member); err != nil { return err } } if err = recountGroupMembersTx(tx, topic.Id); err != nil { return err } if err = updateGroupTopicSubCountTx(tx, 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, "SELECT deletedat FROM im_subscriptions WHERE topic=? AND userid=? FOR UPDATE", 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 = createSubscription(tx, 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 (invitation_id,topic,inviter_user_id,target_user_id,message,status,result,idempotency_key,expires_at,handled_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`, invitationID, topic, decodeUID(inviter), decodeUID(target), message, 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, member); err != nil { return false, err } if err = recountGroupMembersAndBumpVersionTx(tx, topic); err != nil { return false, err } if err = updateGroupTopicSubCountTx(tx, 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, "UPDATE im_subscriptions SET updatedat=?,deletedat=? WHERE topic=? AND userid=? AND deletedat IS NULL", now, now, topic, decodeUID(uid)) if err != nil { return err } if affected, _ := res.RowsAffected(); affected == 0 { return t.ErrNotFound } if _, err = tx.ExecContext(ctx, `UPDATE im_group_members SET member_status=?,left_at=?,version=version+1,updated_at=? WHERE topic=? AND user_id=? AND member_status=?`, t.GroupMemberStatusLeft, now, now, topic, decodeUID(uid), t.GroupMemberStatusActive); err != nil { return err } if err = recountGroupMembersAndBumpVersionTx(tx, topic); err != nil { return err } if err = updateGroupTopicSubCountTx(tx, 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, "UPDATE im_subscriptions SET modegiven=?,modewant=?,updatedat=? WHERE topic=? AND userid=? AND deletedat IS NULL", modeGiven.String(), modeGiven.String(), now, topic, decodeUID(target)) if err != nil { return err } if affected, _ := res.RowsAffected(); affected == 0 { return t.ErrNotFound } res, err = tx.ExecContext(ctx, "UPDATE im_group_members SET role=?,version=version+1,updated_at=? WHERE topic=? AND user_id=? AND member_status=?", role, now, 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, "UPDATE im_subscriptions SET modegiven=?,modewant=?,updatedat=? WHERE topic=? AND userid=? AND deletedat IS NULL", update.mode.String(), update.mode.String(), now, topic, decodeUID(update.uid)) if execErr != nil { return execErr } if affected, _ := res.RowsAffected(); affected == 0 { return t.ErrNotFound } res, execErr = tx.ExecContext(ctx, "UPDATE im_group_members SET role=?,version=version+1,updated_at=? WHERE topic=? AND user_id=? AND member_status=?", update.role, now, 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, "UPDATE im_topics SET owner=?,updatedat=? WHERE name=?", decodeUID(newOwner), now, topic); err != nil { return err } args := []any{decodeUID(newOwner), now, topic} query := "UPDATE im_group_settings SET owner_user_id=?,version=version+1,updated_at=? WHERE 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 topic=? AND group_status=? LIMIT 1),0), EXISTS(SELECT 1 FROM im_group_mutes WHERE topic=? AND user_id=? AND status=? AND (expires_at IS NULL OR expires_at>?))`, topic, t.GroupStatusActive, topic, decodeUID(uid), t.GroupMuteStatusActive, now).Scan(&muteAll, &memberMuted) if err != nil { return false, err } return !muteAll && !memberMuted, nil } func insertGroupSettingsTx(tx *sqlx.Tx, settings *t.GroupSettings) error { settingsID, err := nextSnowflakeID() if err != nil { return err } _, err = tx.Exec(`INSERT INTO im_group_settings (id,topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,created_by,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, settingsID, settings.Topic, decodeUID(settings.OwnerUserID), settings.Name, settings.AvatarFileID, settings.Description, settings.GroupStatus, settings.JoinApproval, settings.MuteAll, settings.MemberUploadFiles, settings.MemberPrivateChat, settings.MemberInvite, settings.MaxMemberCount, settings.MaxAdminCount, settings.MemberCount, settings.Version, decodeUID(settings.CreatedBy), settings.CreatedAt, settings.UpdatedAt) return err } func upsertGroupMemberTx(tx *sqlx.Tx, member *t.GroupMember) error { _, err := tx.Exec(`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=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) return err } func recountGroupMembersTx(tx *sqlx.Tx, topic string) error { _, err := tx.Exec(`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 recountGroupMembersAndBumpVersionTx(tx *sqlx.Tx, topic string) error { _, err := tx.Exec(`UPDATE im_group_settings SET member_count=(SELECT COUNT(*) FROM im_group_members WHERE topic=? AND member_status=?),version=version+1,updated_at=CURRENT_TIMESTAMP(3) WHERE topic=?`, topic, t.GroupMemberStatusActive, topic) return err } func updateGroupTopicSubCountTx(tx *sqlx.Tx, topic string) error { _, err := tx.Exec(`UPDATE im_topics SET subcnt=(SELECT COUNT(*) FROM im_subscriptions WHERE topic IN (?,?) AND deletedat IS NULL) WHERE name=?`, topic, t.GrpToChn(topic), topic) return err } func syncGroupMuteAllTx(tx *sqlx.Tx, topic string, muteAll bool, now time.Time) error { mode := t.ModeCPublic if muteAll { mode = t.ModeCPublic &^ t.ModeWrite } _, err := tx.Exec(`UPDATE im_subscriptions AS s JOIN im_group_members AS gm ON 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, topic string) (*t.GroupSettings, error) { var item t.GroupSettings err := a.db.QueryRowxContext(ctx, "SELECT "+groupSettingsSelectColumns+" 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) 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.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, 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 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 }