From 513509c5e62b975ada3ce6bb9cdd1a0e3d12d1c0 Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 20 Jul 2026 14:50:18 +0800 Subject: [PATCH] feat(cp): notify when relationship capacity is reached --- docs/CP关系服务端开发文档.md | 21 ++++- .../modules/cpnotice/mysql_repository.go | 9 +- .../internal/modules/cpnotice/service.go | 76 ++++++++++++++-- .../user-service/internal/domain/cp/cp.go | 15 +++ .../internal/storage/mysql/cp/repository.go | 91 +++++++++++++++---- 5 files changed, 177 insertions(+), 35 deletions(-) diff --git a/docs/CP关系服务端开发文档.md b/docs/CP关系服务端开发文档.md index 4abdad9f..311241d1 100644 --- a/docs/CP关系服务端开发文档.md +++ b/docs/CP关系服务端开发文档.md @@ -89,10 +89,16 @@ sequenceDiagram R->>R: update heat/rank/command log R-->>F: SendGiftResponse R-->>U: RoomGiftSent MQ - U->>U: create or refresh CP application - U-->>N: UserCPApplicationCreated MQ - N->>IM: A -> B C2C cp_application_created - N->>IM: room cp_application_created, recipient_user_id=B + U->>U: check relationship capacity + alt either user has available capacity + U->>U: create or refresh CP application + U-->>N: UserCPApplicationCreated MQ + N->>IM: A -> B C2C cp_application_created + N->>IM: room cp_application_created, recipient_user_id=B + else either user has reached capacity + U-->>N: UserCPRelationshipCapacityReached MQ + N->>IM: system -> A C2C cp_relationship_capacity_reached + end ``` 处理规则： @@ -101,7 +107,10 @@ sequenceDiagram - 同一个 `RoomGiftSent.event_id` 只能消费一次。 - A 给 B 发送同一类型 CP 礼物时，如果已有 pending 申请，刷新礼物快照和 `expires_at_ms`。 - A 给 B 发送不同类型 CP 礼物时，可以各有一条 pending 申请。 -- 如果任一方已经有 active 关系，申请可以存在，但接受时会被拒绝；已有关系双方互送礼物会增加亲密值。 +- A/B 已经是 active 关系时，双方互送任意礼物只增加亲密值，不重复创建申请。 +- A/B 尚无 active 关系，但任一方该类型 active 关系数量已达上限时，不创建或刷新申请；user-service 同事务写入消费位点和 `UserCPRelationshipCapacityReached` outbox，notice-service 再由腾讯 IM 管理员账号通知送礼人 A。 +- 容量提示 payload 使用 `event_type=cp_relationship_capacity_reached`、`recipient_user_id=A`、`message_text=用户关系已经达到上限`；App 按 `event_id` 去重并展示轻提示。 +- 送礼扣费和房间表现先于异步容量判断完成；容量已满只阻止创建无效申请，不回滚礼物账务。 ## 6. 同意和拒绝 @@ -270,6 +279,7 @@ sequenceDiagram | `cp_application_rejected` | C2C | B 拒绝后，B 发给 A | | `cp_application_rejected` | 房间群 | B 拒绝后，发到来源房间，payload 带 `recipient_user_id=A` | | `cp_relationship_created` | 区域群 | 关系建立后发送区域播报 | +| `cp_relationship_capacity_reached` | 系统 C2C | A 送关系礼物后，A 或 B 的该类型 active 关系已达上限；管理员账号发给 A，payload 带 `message_text=用户关系已经达到上限` | | 无 | 无 | 解除关系成功不发 IM，App 根据解除接口返回刷新 | C2C 使用腾讯云 IM `From_Account` 表达会话方向：申请为 A，拒绝为 B；同意会发 `B -> A` 和 `A -> B` 两条。区域 IM 群 ID 使用 `hy__bc_r_v2_`（v1 旧群 `hy__bc_r_` 仅保留过渡期双发），本地测试可通过 `group_id_prefix` 加前缀。 @@ -296,3 +306,4 @@ C2C 使用腾讯云 IM `From_Account` 表达会话方向：申请为 A，拒绝 - active 关系双方互送任意礼物，亲密值按 `heat_value` 增加。 - active 关系解除会扣除后台配置的金币费用，扣费失败不改变关系状态。 - 同意、拒绝、关系成立的 IM 都写入 `notice_delivery_events`，失败可重试。 +- 关系容量已满后再次送关系礼物，会写 `relationship_capacity_reached` 消费位点，不生成 pending 申请，并把 `cp_relationship_capacity_reached` 系统 IM 投递给送礼人。 diff --git a/services/notice-service/internal/modules/cpnotice/mysql_repository.go b/services/notice-service/internal/modules/cpnotice/mysql_repository.go index 364ef07a..e76da795 100644 --- a/services/notice-service/internal/modules/cpnotice/mysql_repository.go +++ b/services/notice-service/internal/modules/cpnotice/mysql_repository.go @@ -192,7 +192,7 @@ func validateDelivery(delivery CPDelivery) error { if isTencentIMC2CChannel(delivery.Channel) && delivery.TargetUserID <= 0 { return xerr.New(xerr.InvalidArgument, "cp c2c delivery target_user_id is required") } - if isTencentIMC2CChannel(delivery.Channel) && delivery.SenderUserID <= 0 { + if isUserAuthoredTencentIMC2CChannel(delivery.Channel) && delivery.SenderUserID <= 0 { return xerr.New(xerr.InvalidArgument, "cp c2c delivery sender_user_id is required") } if (delivery.Channel == channelTencentIMRoom || delivery.Channel == channelTencentIMRoomSender || delivery.Channel == channelTencentIMRegion) && strings.TrimSpace(delivery.GroupID) == "" { @@ -203,13 +203,18 @@ func validateDelivery(delivery CPDelivery) error { func isTencentIMC2CChannel(channel string) bool { switch channel { - case channelTencentIMC2C, channelTencentIMC2CAcceptor: + case channelTencentIMC2C, channelTencentIMC2CAcceptor, channelTencentIMSystemC2C: return true default: return false } } +func isUserAuthoredTencentIMC2CChannel(channel string) bool { + // 系统 C2C 必须让 REST client 使用管理员账号；只有关系双方代发的 C2C 才要求 sender_user_id。 + return channel == channelTencentIMC2C || channel == channelTencentIMC2CAcceptor +} + func truncateError(value string) string { value = strings.TrimSpace(value) if len(value) <= 512 { diff --git a/services/notice-service/internal/modules/cpnotice/service.go b/services/notice-service/internal/modules/cpnotice/service.go index abe70d60..c76d9396 100644 --- a/services/notice-service/internal/modules/cpnotice/service.go +++ b/services/notice-service/internal/modules/cpnotice/service.go @@ -17,21 +17,24 @@ import ( ) const ( - eventApplicationCreated = "UserCPApplicationCreated" - eventApplicationAccepted = "UserCPApplicationAccepted" - eventApplicationRejected = "UserCPApplicationRejected" - eventRelationshipCreated = "UserCPRelationshipCreated" + eventApplicationCreated = "UserCPApplicationCreated" + eventApplicationAccepted = "UserCPApplicationAccepted" + eventApplicationRejected = "UserCPApplicationRejected" + eventRelationshipCreated = "UserCPRelationshipCreated" + eventRelationshipCapacityReached = "UserCPRelationshipCapacityReached" channelTencentIMC2C = "tencent_im_c2c" channelTencentIMC2CAcceptor = "tencent_im_c2c_acceptor" + channelTencentIMSystemC2C = "tencent_im_system_c2c" channelTencentIMRoom = "tencent_im_room" channelTencentIMRoomSender = "tencent_im_room_sender" channelTencentIMRegion = "tencent_im_region" - noticeTypeApplicationCreated = "cp_application_created" - noticeTypeApplicationAccepted = "cp_application_accepted" - noticeTypeApplicationRejected = "cp_application_rejected" - noticeTypeRelationshipCreated = "cp_relationship_created" + noticeTypeApplicationCreated = "cp_application_created" + noticeTypeApplicationAccepted = "cp_application_accepted" + noticeTypeApplicationRejected = "cp_application_rejected" + noticeTypeRelationshipCreated = "cp_relationship_created" + noticeTypeRelationshipCapacityReached = "cp_relationship_capacity_reached" ) // Repository owns notice_delivery_events for CP IM deliveries; user-service remains the CP fact owner. @@ -116,6 +119,16 @@ func (s *Service) publishDelivery(ctx context.Context, delivery CPDelivery, opti Ext: "cp_relation_notice", PayloadJSON: delivery.PayloadJSON, }) + case channelTencentIMSystemC2C: + // FromAccount 留空时腾讯 IM client 会使用配置的管理员账号；容量告警属于系统事实， + // 不能伪装成收礼人发出的 C2C 消息，也不能污染双方的关系会话方向。 + err = s.publisher.PublishUserCustomMessage(publishCtx, tencentim.CustomUserMessage{ + ToAccount: tencentim.FormatUserID(delivery.TargetUserID), + EventID: delivery.EventID, + Desc: delivery.NoticeType, + Ext: "system_notice", + PayloadJSON: delivery.PayloadJSON, + }) case channelTencentIMRoom, channelTencentIMRoomSender: err = s.publisher.PublishGroupCustomMessage(publishCtx, tencentim.CustomGroupMessage{ GroupID: delivery.GroupID, @@ -235,7 +248,7 @@ func (s *Service) markFailed(ctx context.Context, delivery CPDelivery, options C func isCPNoticeEvent(eventType string) bool { switch eventType { - case eventApplicationCreated, eventApplicationAccepted, eventApplicationRejected, eventRelationshipCreated: + case eventApplicationCreated, eventApplicationAccepted, eventApplicationRejected, eventRelationshipCreated, eventRelationshipCapacityReached: return true default: return false @@ -333,11 +346,56 @@ func deliveriesFromMessage(message usermq.UserOutboxMessage, groupIDPrefix strin return nil, err } return relationshipCreatedDelivery(appCode, message, application, relationshipID, groupID) + case eventRelationshipCapacityReached: + payload, targetUserID, err := cpRelationshipCapacityReachedPayload(message, root) + if err != nil { + return nil, err + } + return []CPDelivery{{ + AppCode: appCode, + EventID: message.EventID, + EventType: message.EventType, + Channel: channelTencentIMSystemC2C, + NoticeType: noticeTypeRelationshipCapacityReached, + TargetUserID: targetUserID, + PayloadJSON: mustJSON(payload), + CreatedAtMS: message.OccurredAtMS, + }}, nil default: return nil, nil } } +func cpRelationshipCapacityReachedPayload(message usermq.UserOutboxMessage, root map[string]any) (map[string]any, int64, error) { + senderUserID := int64Value(root, "sender_user_id", "SenderUserID") + targetUserID := int64Value(root, "target_user_id", "TargetUserID") + limitedUserID := int64Value(root, "limited_user_id", "LimitedUserID") + relationType := stringValue(root, "relation_type", "RelationType") + messageText := stringValue(root, "message_text", "MessageText") + if senderUserID <= 0 || targetUserID <= 0 || limitedUserID <= 0 || relationType == "" || messageText == "" { + return nil, 0, fmt.Errorf("cp relationship capacity payload is incomplete") + } + // 最终 IM 使用字符串承载 user_id，避免 Dart/JavaScript 客户端经过 JSON number 时丢失 BIGINT 精度。 + payload := map[string]any{ + "event_id": message.EventID, + "event_type": noticeTypeRelationshipCapacityReached, + "source_event_type": message.EventType, + "app_code": appcode.Normalize(message.AppCode), + "relation_type": relationType, + "sender_user_id": strconv.FormatInt(senderUserID, 10), + "target_user_id": strconv.FormatInt(targetUserID, 10), + "limited_user_id": strconv.FormatInt(limitedUserID, 10), + "recipient_user_id": strconv.FormatInt(senderUserID, 10), + "current_count": int64Value(root, "current_count", "CurrentCount"), + "max_count": int64Value(root, "max_count", "MaxCount"), + "room_id": stringValue(root, "room_id", "RoomID"), + "source_room_event_id": stringValue(root, "source_room_event_id", "SourceRoomEventID"), + "message_text": messageText, + "source_created_at_ms": message.OccurredAtMS, + } + return payload, senderUserID, nil +} + func applicationDecisionDeliveries(appCode string, message usermq.UserOutboxMessage, application cpApplication, payload map[string]any, noticeType string) []CPDelivery { // 同意/拒绝结果必须同步覆盖两个入口： // 1. A 在私聊里看到 C2C 结果； diff --git a/services/user-service/internal/domain/cp/cp.go b/services/user-service/internal/domain/cp/cp.go index 128ebb5c..e14b07cd 100644 --- a/services/user-service/internal/domain/cp/cp.go +++ b/services/user-service/internal/domain/cp/cp.go @@ -25,6 +25,7 @@ const ( EventTypeApplicationAccepted = "UserCPApplicationAccepted" EventTypeApplicationRejected = "UserCPApplicationRejected" EventTypeRelationshipCreated = "UserCPRelationshipCreated" + EventTypeRelationshipCapacityReached = "UserCPRelationshipCapacityReached" EventTypeRelationshipIntimacyChanged = "UserCPRelationshipIntimacyChanged" EventTypeRelationshipLevelUp = "UserCPRelationshipLevelUp" EventTypeRelationshipEnded = "UserCPRelationshipEnded" @@ -203,3 +204,17 @@ type ConsumeResult struct { Application Application Relationship Relationship } + +// RelationshipCapacityReachedNotice 是关系礼物已扣费、但任一方该类型 active 关系名额已满时的通知事实。 +// user-service 只固化判断结果；notice-service 再把它转换为发给送礼人的系统 IM。 +type RelationshipCapacityReachedNotice struct { + RelationType string `json:"relation_type"` + SenderUserID int64 `json:"sender_user_id"` + TargetUserID int64 `json:"target_user_id"` + LimitedUserID int64 `json:"limited_user_id"` + CurrentCount int64 `json:"current_count"` + MaxCount int32 `json:"max_count"` + RoomID string `json:"room_id"` + SourceRoomEventID string `json:"source_room_event_id"` + MessageText string `json:"message_text"` +} diff --git a/services/user-service/internal/storage/mysql/cp/repository.go b/services/user-service/internal/storage/mysql/cp/repository.go index 5bc02179..ce408d71 100644 --- a/services/user-service/internal/storage/mysql/cp/repository.go +++ b/services/user-service/internal/storage/mysql/cp/repository.go @@ -17,6 +17,8 @@ import ( const formationGiftCoinThreshold = int64(5000) +const relationshipCapacityReachedMessage = "用户关系已经达到上限" + // Repository 是 CP 关系在 user-service MySQL 上的 owner 存储。 type Repository struct { db *sql.DB @@ -712,6 +714,33 @@ func (r *Repository) ConsumeGiftEvent(ctx context.Context, event cpdomain.GiftEv } return cpdomain.ConsumeResult{Consumed: true}, tx.Commit() } + // 关系礼物扣费和房间表现已经在上游完成，这里只能根据 user-service 的关系事实判断申请是否仍有意义。 + // 任一方该类型名额已满时不再制造一条注定无法同意的 pending 申请，而是同事务记录消费位点和系统通知 outbox； + // notice-service 最终只通知送礼人，避免目标用户收到不可操作的申请卡片。 + capacity, err := relationCapacityTx(ctx, tx, appCode, relationType, event.SenderUserID, event.TargetUserID) + if err != nil { + return cpdomain.ConsumeResult{}, err + } + if capacity.LimitedUserID > 0 { + if err := insertConsumptionTx(ctx, tx, appCode, event.EventID, "skipped", "", "", "relationship_capacity_reached", event.GiftValue, event.OccurredAtMS, nowMs); err != nil { + return cpdomain.ConsumeResult{}, err + } + notice := cpdomain.RelationshipCapacityReachedNotice{ + RelationType: relationType, + SenderUserID: event.SenderUserID, + TargetUserID: event.TargetUserID, + LimitedUserID: capacity.LimitedUserID, + CurrentCount: capacity.CurrentCount, + MaxCount: capacity.MaxCount, + RoomID: event.RoomID, + SourceRoomEventID: event.EventID, + MessageText: relationshipCapacityReachedMessage, + } + if err := insertOutboxTx(ctx, tx, appCode, cpdomain.EventTypeRelationshipCapacityReached, "cp_relationship_capacity", event.SenderUserID, notice, nowMs); err != nil { + return cpdomain.ConsumeResult{}, err + } + return cpdomain.ConsumeResult{Consumed: true}, tx.Commit() + } application, err := upsertApplicationFromGiftTx(ctx, tx, appCode, event, relationType, nowMs) if err != nil { return cpdomain.ConsumeResult{}, err @@ -1085,40 +1114,64 @@ func ensurePairHasNoActiveRelationshipTx(ctx context.Context, tx *sql.Tx, appCod return nil } -func assertRelationCapacityTx(ctx context.Context, tx *sql.Tx, appCode string, relationType string, userAID int64, userBID int64, actorUserID int64) error { +type relationCapacity struct { + LimitedUserID int64 + CurrentCount int64 + MaxCount int32 +} + +func relationCapacityTx(ctx context.Context, tx *sql.Tx, appCode string, relationType string, userAID int64, userBID int64) (relationCapacity, error) { if normalizeRelationType(relationType) == "" { - return xerr.New(xerr.Conflict, "cp relation type is disabled") + return relationCapacity{}, xerr.New(xerr.Conflict, "cp relation type is disabled") } maxCount, err := relationMaxCountTx(ctx, tx, appCode, relationType) if err != nil { - return err + return relationCapacity{}, err } for _, userID := range []int64{userAID, userBID} { // 容量按用户和关系类型计算；同一对用户互斥由 pair 锁保证，不同对象可以同时拥有兄弟/姐妹关系。 + // user_a/user_b 分支分别命中 idx_user_cp_rel_user_a / idx_user_cp_rel_user_b，避免 OR 条件退化为大范围扫描。 total, err := countRows(ctx, tx, ` SELECT COUNT(*) - FROM user_cp_relationships - WHERE app_code = ? AND status = ? - AND relation_type = ? - AND (user_a_id = ? OR user_b_id = ?)`, - appCode, cpdomain.RelationshipStatusActive, relationType, userID, userID, + FROM ( + SELECT relationship_id + FROM user_cp_relationships FORCE INDEX (idx_user_cp_rel_user_a) + WHERE app_code = ? AND user_a_id = ? AND status = ? AND relation_type = ? + UNION ALL + SELECT relationship_id + FROM user_cp_relationships FORCE INDEX (idx_user_cp_rel_user_b) + WHERE app_code = ? AND user_b_id = ? AND status = ? AND relation_type = ? + ) active_relationships`, + appCode, userID, cpdomain.RelationshipStatusActive, relationType, + appCode, userID, cpdomain.RelationshipStatusActive, relationType, ) if err != nil { - return err + return relationCapacity{}, err } if total >= int64(maxCount) { - // CP 固定一人一个；同意按钮的操作者需要知道是自己还是申请另一方已占用名额。 - // 兄弟/姐妹由后台配置多名额，继续使用通用容量错误，避免把非 CP 关系误称为 CP。 - if normalizeRelationType(relationType) == cpdomain.RelationTypeCP { - if userID == actorUserID { - return xerr.New(xerr.CPAlreadyExistsSelf, "current user already has an active cp relationship") - } - return xerr.New(xerr.CPAlreadyExistsOther, "other party already has an active cp relationship") - } - return xerr.New(xerr.Conflict, "user cp relation count reaches limit") + return relationCapacity{LimitedUserID: userID, CurrentCount: total, MaxCount: maxCount}, nil } } - return nil + return relationCapacity{MaxCount: maxCount}, nil +} + +func assertRelationCapacityTx(ctx context.Context, tx *sql.Tx, appCode string, relationType string, userAID int64, userBID int64, actorUserID int64) error { + capacity, err := relationCapacityTx(ctx, tx, appCode, relationType, userAID, userBID) + if err != nil { + return err + } + if capacity.LimitedUserID <= 0 { + return nil + } + // CP 固定一人一个；同意按钮的操作者需要知道是自己还是申请另一方已占用名额。 + // 兄弟/姐妹由后台配置多名额，继续使用通用容量错误，避免把非 CP 关系误称为 CP。 + if normalizeRelationType(relationType) == cpdomain.RelationTypeCP { + if capacity.LimitedUserID == actorUserID { + return xerr.New(xerr.CPAlreadyExistsSelf, "current user already has an active cp relationship") + } + return xerr.New(xerr.CPAlreadyExistsOther, "other party already has an active cp relationship") + } + return xerr.New(xerr.Conflict, "user cp relation count reaches limit") } func relationMaxCountTx(ctx context.Context, tx *sql.Tx, appCode string, relationType string) (int32, error) {