From 1c45fdc65b9df459c7e014f0305628cd115f1a66 Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 20 Jul 2026 18:30:56 +0800 Subject: [PATCH] fix(user): make agency invitations retry-safe --- .../internal/storage/mysql/host/admin.go | 14 ++++++++++++ .../storage/mysql/host/invitations.go | 22 +++++++++++++++++++ .../internal/storage/mysql/host/queries.go | 17 ++++++++++++++ 3 files changed, 53 insertions(+) diff --git a/services/user-service/internal/storage/mysql/host/admin.go b/services/user-service/internal/storage/mysql/host/admin.go index 70868b12..92d2f51a 100644 --- a/services/user-service/internal/storage/mysql/host/admin.go +++ b/services/user-service/internal/storage/mysql/host/admin.go @@ -314,6 +314,20 @@ func (r *Repository) CreateAgency(ctx context.Context, command hostservice.Creat } else if ok { return hostdomain.CreateAgencyResult{}, xerr.New(xerr.Conflict, "owner already has active agency") } + // 后台直建 Agency 是对角色归属的最终确认；目标先前收到的 Agency 邀请已经不可再接受。 + // 必须在同一事务关闭 pending 消息，否则消息中心会长期展示一个点击后必然冲突的旧邀请。 + if pending, ok, err := queryPendingRoleInvitationByTarget(ctx, tx, hostdomain.InvitationTypeAgency, command.OwnerUserID, true); err != nil { + return hostdomain.CreateAgencyResult{}, err + } else if ok { + pending.Status = hostdomain.InvitationStatusCancelled + pending.ProcessedByUserID = command.AdminUserID + pending.ProcessReason = "superseded by admin agency creation" + pending.ProcessedAtMs = command.NowMs + pending.UpdatedAtMs = command.NowMs + if err := updateRoleInvitationProcessed(ctx, tx, pending); err != nil { + return hostdomain.CreateAgencyResult{}, err + } + } activeMembership, hasActiveMembership, err := queryActiveMembershipByHost(ctx, tx, command.OwnerUserID, true) if err != nil { return hostdomain.CreateAgencyResult{}, err diff --git a/services/user-service/internal/storage/mysql/host/invitations.go b/services/user-service/internal/storage/mysql/host/invitations.go index 73a26ad1..614649d7 100644 --- a/services/user-service/internal/storage/mysql/host/invitations.go +++ b/services/user-service/internal/storage/mysql/host/invitations.go @@ -54,6 +54,28 @@ func (r *Repository) InviteAgency(ctx context.Context, command hostservice.Invit } else if ok { return hostdomain.RoleInvitation{}, xerr.New(xerr.Conflict, "target already owns active agency") } + // H5 在首个响应丢失、弹窗重新打开等情况下可能为同一目标生成新的 command_id。 + // 同一 BD 的 pending 邀请属于同一业务动作，记录新命令到既有结果并按成功回放；其他 BD 仍不能覆盖目标已收到的邀请。 + if pending, ok, err := queryPendingRoleInvitationByTarget(ctx, tx, hostdomain.InvitationTypeAgency, command.TargetUserID, true); err != nil { + return hostdomain.RoleInvitation{}, err + } else if ok { + if pending.InviterBDUserID != inviter.UserID { + return hostdomain.RoleInvitation{}, xerr.New(xerr.Conflict, "target already has pending agency invitation") + } + if err := insertCommandResult(ctx, tx, hostdomain.CommandResult{ + CommandID: command.CommandID, + CommandType: hostdomain.CommandTypeInviteAgency, + ResultType: hostdomain.ResultTypeInvitation, + ResultID: pending.InvitationID, + CreatedAtMs: command.NowMs, + }); err != nil { + return hostdomain.RoleInvitation{}, mapHostDuplicateError(err) + } + if err := tx.Commit(); err != nil { + return hostdomain.RoleInvitation{}, err + } + return pending, nil + } // BD 发出的 Agency 邀请要记录直接上级 BD 和负责人链路； // 如果本次以负责人身份邀请，则直接 BD 和负责人都指向 Leader 自己。 diff --git a/services/user-service/internal/storage/mysql/host/queries.go b/services/user-service/internal/storage/mysql/host/queries.go index 657272ff..477a0b04 100644 --- a/services/user-service/internal/storage/mysql/host/queries.go +++ b/services/user-service/internal/storage/mysql/host/queries.go @@ -602,6 +602,23 @@ func (r *Repository) ListRoleInvitations(ctx context.Context, command hostservic return queryRoleInvitations(ctx, r.db, query, args...) } +// queryPendingRoleInvitationByTarget 通过 pending_target_key 唯一索引读取目标当前邀请。 +// 邀请创建事务传入 forUpdate=true，把“检查 pending + 创建/回放邀请”串行化，避免连续点击用不同 command_id 撞成泛化 409。 +func queryPendingRoleInvitationByTarget(ctx context.Context, q sqlQueryer, invitationType string, targetUserID int64, forUpdate bool) (hostdomain.RoleInvitation, bool, error) { + query := fmt.Sprintf(`SELECT %s FROM role_invitations WHERE app_code = ? AND pending_target_key = ?`, roleInvitationColumns) + if forUpdate { + query += " FOR UPDATE" + } + invitation, err := scanRoleInvitation(q.QueryRowContext(ctx, query, appcode.FromContext(ctx), fmt.Sprintf("%s:%d", invitationType, targetUserID))) + if err == sql.ErrNoRows { + return hostdomain.RoleInvitation{}, false, nil + } + if err != nil { + return hostdomain.RoleInvitation{}, false, err + } + return invitation, true, nil +} + // ListAgencyMembers 读取 Agency 成员关系列表。 func (r *Repository) ListAgencyMembers(ctx context.Context, agencyID int64, status string) ([]hostdomain.AgencyMembership, error) { query := fmt.Sprintf(`SELECT %s FROM agency_memberships WHERE app_code = ? AND agency_id = ?`, agencyMembershipColumns)