2026-07-04 01:53:10 +08:00

517 lines
20 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

// Package cpnotice consumes user-service CP outbox facts and delivers the IM side effects.
package cpnotice
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"strconv"
"time"
"hyapp/pkg/appcode"
"hyapp/pkg/imgroup"
"hyapp/pkg/logx"
"hyapp/pkg/tencentim"
"hyapp/pkg/usermq"
)
const (
eventApplicationCreated = "UserCPApplicationCreated"
eventApplicationAccepted = "UserCPApplicationAccepted"
eventApplicationRejected = "UserCPApplicationRejected"
eventRelationshipCreated = "UserCPRelationshipCreated"
channelTencentIMC2C = "tencent_im_c2c"
channelTencentIMC2CAcceptor = "tencent_im_c2c_acceptor"
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"
)
// Repository owns notice_delivery_events for CP IM deliveries; user-service remains the CP fact owner.
type Repository interface {
ClaimCPDelivery(ctx context.Context, workerID string, delivery CPDelivery, lockTTL time.Duration) (CPDelivery, bool, error)
MarkCPDelivered(ctx context.Context, delivery CPDelivery, deliveredPayload []byte, nowMs int64) error
MarkCPRetryable(ctx context.Context, delivery CPDelivery, retryCount int, nextRetryAtMS int64, lastErr string, nowMs int64) error
MarkCPFailed(ctx context.Context, delivery CPDelivery, retryCount int, lastErr string, nowMs int64) error
}
// Publisher is the Tencent IM boundary needed by CP notices.
type Publisher interface {
PublishUserCustomMessage(ctx context.Context, message tencentim.CustomUserMessage) error
PublishGroupCustomMessage(ctx context.Context, message tencentim.CustomGroupMessage) error
EnsureGroup(ctx context.Context, spec tencentim.GroupSpec) error
}
// Config carries process-level delivery settings that must match gateway broadcast group generation.
type Config struct {
NodeID string
GroupIDPrefix string
}
// Service maps CP facts to concrete IM messages and stores per-channel delivery state.
type Service struct {
cfg Config
repository Repository
publisher Publisher
}
// New creates CP notice service.
func New(cfg Config, repository Repository, publisher Publisher) *Service {
return &Service{cfg: cfg, repository: repository, publisher: publisher}
}
// ProcessUserOutboxMessage handles one user_outbox MQ message. Unknown events are acknowledged so that unrelated
// user-service facts can share the same topic without notice-service poisoning the consumer group.
func (s *Service) ProcessUserOutboxMessage(ctx context.Context, message usermq.UserOutboxMessage, options CPNoticeWorkerOptions) (bool, error) {
if !isCPNoticeEvent(message.EventType) {
return false, nil
}
options = normalizeCPNoticeWorkerOptions(options, s.cfg.NodeID)
if s.repository == nil {
return true, fmt.Errorf("notice repository is not configured")
}
if s.publisher == nil {
return true, fmt.Errorf("notice publisher is not configured")
}
deliveries, err := deliveriesFromMessage(message, s.cfg.GroupIDPrefix)
if err != nil {
return true, err
}
for _, delivery := range deliveries {
claimedDelivery, claimed, err := s.repository.ClaimCPDelivery(ctx, options.WorkerID, delivery, options.LockTTL)
if err != nil || !claimed {
return true, err
}
if err := s.publishDelivery(ctx, claimedDelivery, options); err != nil {
return true, err
}
}
return true, nil
}
func (s *Service) publishDelivery(ctx context.Context, delivery CPDelivery, options CPNoticeWorkerOptions) error {
publishCtx, cancel := context.WithTimeout(ctx, options.PublishTimeout)
defer cancel()
var err error
switch delivery.Channel {
case channelTencentIMC2C, channelTencentIMC2CAcceptor:
err = s.publisher.PublishUserCustomMessage(publishCtx, tencentim.CustomUserMessage{
ToAccount: tencentim.FormatUserID(delivery.TargetUserID),
FromAccount: tencentim.FormatUserID(delivery.SenderUserID),
EventID: delivery.EventID,
Desc: delivery.NoticeType,
Ext: "cp_relation_notice",
PayloadJSON: delivery.PayloadJSON,
})
case channelTencentIMRoom, channelTencentIMRoomSender:
err = s.publisher.PublishGroupCustomMessage(publishCtx, tencentim.CustomGroupMessage{
GroupID: delivery.GroupID,
EventID: delivery.EventID,
Desc: delivery.NoticeType,
Ext: "cp_room_notice",
PayloadJSON: delivery.PayloadJSON,
})
case channelTencentIMRegion:
// 区域群是长期播报群;发送前幂等创建,避免本地真实测试或新区域缺群导致关系主流程事实无法通知。
if ensureErr := s.publisher.EnsureGroup(publishCtx, s.regionGroupSpec(delivery.GroupID)); ensureErr != nil {
err = ensureErr
break
}
err = s.publisher.PublishGroupCustomMessage(publishCtx, tencentim.CustomGroupMessage{
GroupID: delivery.GroupID,
EventID: delivery.EventID,
Desc: delivery.NoticeType,
Ext: "cp_region_broadcast",
PayloadJSON: delivery.PayloadJSON,
})
if err == nil {
s.dualSendLegacyRegion(publishCtx, delivery)
}
default:
err = fmt.Errorf("unsupported cp notice channel %q", delivery.Channel)
}
nowMs := time.Now().UnixMilli()
if err == nil {
if markErr := s.repository.MarkCPDelivered(ctx, delivery, delivery.PayloadJSON, nowMs); markErr != nil {
return markErr
}
logx.Info(ctx, "notice_cp_delivered",
slog.String("event_id", delivery.EventID),
slog.String("app_code", delivery.AppCode),
slog.String("notice_type", delivery.NoticeType),
slog.String("channel", delivery.Channel),
slog.Int64("target_user_id", delivery.TargetUserID),
slog.String("group_id", delivery.GroupID),
)
return nil
}
return s.markFailed(ctx, delivery, options, err)
}
// regionGroupSpec 按群 ID 版本决定建群参数v2 区域群是 AVChatRoomChatRoom 套餐成员上限
// 曾在线上触发 10038 拒绝加群),迁移前入队的存量 v1 投递记录仍按 ChatRoom 幂等创建。
func (s *Service) regionGroupSpec(groupID string) tencentim.GroupSpec {
parsed := imgroup.ParseWithPrefix(groupID, s.cfg.GroupIDPrefix)
if parsed.Kind == imgroup.KindRegionBroadcast {
if legacyID, err := imgroup.LegacyRegionBroadcastGroupIDWithPrefix(s.cfg.GroupIDPrefix, parsed.AppCode, parsed.RegionID); err == nil && legacyID != groupID {
return tencentim.GroupSpec{
GroupID: groupID,
Name: groupID,
Type: tencentim.BroadcastGroupType,
}
}
}
return tencentim.GroupSpec{
GroupID: groupID,
Name: groupID,
Type: tencentim.DefaultGroupType,
ApplyJoinOption: "FreeAccess",
}
}
// dualSendLegacyRegion 在区域播报群 v1→v2 过渡期内,把已成功投递到 v2 群的 CP 播报补发到 v1 旧群,
// 覆盖尚未重新拉取 join 列表的老客户端。补发尽力而为,失败不影响主投递状态,与 activity 播报双发同一模式。
func (s *Service) dualSendLegacyRegion(ctx context.Context, delivery CPDelivery) {
parsed := imgroup.ParseWithPrefix(delivery.GroupID, s.cfg.GroupIDPrefix)
if parsed.Kind != imgroup.KindRegionBroadcast {
return
}
legacyID, err := imgroup.LegacyRegionBroadcastGroupIDWithPrefix(s.cfg.GroupIDPrefix, parsed.AppCode, parsed.RegionID)
if err != nil || legacyID == delivery.GroupID {
return
}
_ = s.publisher.PublishGroupCustomMessage(ctx, tencentim.CustomGroupMessage{
GroupID: legacyID,
EventID: delivery.EventID + "_legacy",
Desc: delivery.NoticeType,
Ext: "cp_region_broadcast",
PayloadJSON: delivery.PayloadJSON,
})
}
func (s *Service) markFailed(ctx context.Context, delivery CPDelivery, options CPNoticeWorkerOptions, cause error) error {
nowMs := time.Now().UnixMilli()
nextRetryCount := delivery.RetryCount + 1
if nextRetryCount >= options.MaxRetryCount {
if err := s.repository.MarkCPFailed(ctx, delivery, nextRetryCount, cause.Error(), nowMs); err != nil {
return err
}
logx.Error(ctx, "notice_cp_dead_letter", cause,
slog.String("event_id", delivery.EventID),
slog.String("app_code", delivery.AppCode),
slog.String("notice_type", delivery.NoticeType),
slog.String("channel", delivery.Channel),
slog.Int("retry_count", nextRetryCount),
)
return nil
}
nextRetryAtMS := nowMs + cpNoticeBackoff(nextRetryCount, options).Milliseconds()
if err := s.repository.MarkCPRetryable(ctx, delivery, nextRetryCount, nextRetryAtMS, cause.Error(), nowMs); err != nil {
return err
}
logx.Error(ctx, "notice_cp_retryable", cause,
slog.String("event_id", delivery.EventID),
slog.String("app_code", delivery.AppCode),
slog.String("notice_type", delivery.NoticeType),
slog.String("channel", delivery.Channel),
slog.Int("retry_count", nextRetryCount),
slog.Int64("next_retry_at_ms", nextRetryAtMS),
)
return nil
}
func isCPNoticeEvent(eventType string) bool {
switch eventType {
case eventApplicationCreated, eventApplicationAccepted, eventApplicationRejected, eventRelationshipCreated:
return true
default:
return false
}
}
func deliveriesFromMessage(message usermq.UserOutboxMessage, groupIDPrefix string) ([]CPDelivery, error) {
appCode := appcode.Normalize(message.AppCode)
root, err := decodeObject(message.PayloadJSON)
if err != nil {
return nil, err
}
switch message.EventType {
case eventApplicationCreated:
application, err := applicationFromObject(root)
if err != nil {
return nil, err
}
basePayload, err := cpApplicationPayload(message, application, noticeTypeApplicationCreated, "pending")
if err != nil {
return nil, err
}
roomPayload := cloneJSONMap(basePayload)
roomPayload["recipient_user_id"] = strconv.FormatInt(application.Target.UserID, 10)
requesterRoomPayload := cloneJSONMap(basePayload)
requesterRoomPayload["recipient_user_id"] = strconv.FormatInt(application.Requester.UserID, 10)
return []CPDelivery{
{
AppCode: appCode,
EventID: message.EventID,
EventType: message.EventType,
Channel: channelTencentIMC2C,
NoticeType: noticeTypeApplicationCreated,
SenderUserID: application.Requester.UserID,
TargetUserID: application.Target.UserID,
PayloadJSON: mustJSON(basePayload),
CreatedAtMS: message.OccurredAtMS,
},
{
AppCode: appCode,
EventID: message.EventID,
EventType: message.EventType,
Channel: channelTencentIMRoom,
NoticeType: noticeTypeApplicationCreated,
TargetUserID: application.Target.UserID,
GroupID: application.RoomID,
PayloadJSON: mustJSON(roomPayload),
CreatedAtMS: message.OccurredAtMS,
},
{
// 创建申请时 target 需要接收态按钮requester 也需要同一房间上下文里的等待态弹窗。
// notice_delivery_events 的主键包含 channel同一 event_id 不能复用 tencent_im_room
// 因此发起人房间通知用独立内部 channel发布边界仍走腾讯 IM 群自定义消息。
AppCode: appCode,
EventID: message.EventID,
EventType: message.EventType,
Channel: channelTencentIMRoomSender,
NoticeType: noticeTypeApplicationCreated,
TargetUserID: application.Requester.UserID,
GroupID: application.RoomID,
PayloadJSON: mustJSON(requesterRoomPayload),
CreatedAtMS: message.OccurredAtMS,
},
}, nil
case eventApplicationAccepted:
application, err := applicationFromObject(root)
if err != nil {
return nil, err
}
payload, err := cpApplicationPayload(message, application, noticeTypeApplicationAccepted, "accepted")
if err != nil {
return nil, err
}
return applicationAcceptedDeliveries(appCode, message, application, payload), nil
case eventApplicationRejected:
application, err := applicationFromObject(firstObject(root, "application", "Application"))
if err != nil {
return nil, err
}
payload, err := cpApplicationPayload(message, application, noticeTypeApplicationRejected, "rejected")
if err != nil {
return nil, err
}
payload["reason"] = stringValue(root, "reason", "Reason")
return applicationDecisionDeliveries(appCode, message, application, payload, noticeTypeApplicationRejected), nil
case eventRelationshipCreated:
application, err := applicationFromObject(firstObject(root, "application", "Application"))
if err != nil {
return nil, err
}
relationship := firstObject(root, "relationship", "Relationship")
relationshipID := stringValue(relationship, "relationship_id", "RelationshipID")
groupID, err := imgroup.RegionBroadcastGroupIDWithPrefix(groupIDPrefix, appCode, application.RoomRegionID)
if err != nil {
return nil, err
}
return relationshipCreatedDelivery(appCode, message, application, relationshipID, groupID)
default:
return nil, nil
}
}
func applicationDecisionDeliveries(appCode string, message usermq.UserOutboxMessage, application cpApplication, payload map[string]any, noticeType string) []CPDelivery {
// 同意/拒绝结果必须同步覆盖两个入口:
// 1. A 在私聊里看到 C2C 结果;
// 2. A 在申请来源房间上下文看到房间 IM客户端用 recipient_user_id 只给 A 展示按钮态变化。
deliveries := []CPDelivery{{
AppCode: appCode,
EventID: message.EventID,
EventType: message.EventType,
Channel: channelTencentIMC2C,
NoticeType: noticeType,
SenderUserID: application.Target.UserID,
TargetUserID: application.Requester.UserID,
PayloadJSON: mustJSON(payload),
CreatedAtMS: message.OccurredAtMS,
}}
if application.RoomID == "" {
return deliveries
}
roomPayload := cloneJSONMap(payload)
roomPayload["recipient_user_id"] = strconv.FormatInt(application.Requester.UserID, 10)
deliveries = append(deliveries, CPDelivery{
AppCode: appCode,
EventID: message.EventID,
EventType: message.EventType,
Channel: channelTencentIMRoom,
NoticeType: noticeType,
TargetUserID: application.Requester.UserID,
GroupID: application.RoomID,
PayloadJSON: mustJSON(roomPayload),
CreatedAtMS: message.OccurredAtMS,
})
return deliveries
}
func applicationAcceptedDeliveries(appCode string, message usermq.UserOutboxMessage, application cpApplication, payload map[string]any) []CPDelivery {
// 同意成功后双方都要在 C2C 会话里看到一条消息notice_delivery_events 的幂等键只有 event_id + channel
// 所以第二条 C2C 使用独立内部 channel但 publishDelivery 仍按腾讯 IM 单聊投递。
requesterPayload := cloneJSONMap(payload)
requesterPayload["c2c_notice_role"] = "accepted_to_requester"
requesterPayload["message_text"] = "我同意了你"
acceptorPayload := cloneJSONMap(payload)
acceptorPayload["c2c_notice_role"] = "accepted_to_acceptor"
acceptorPayload["message_text"] = "对方已同意你的申请"
deliveries := []CPDelivery{
{
AppCode: appCode,
EventID: message.EventID,
EventType: message.EventType,
Channel: channelTencentIMC2C,
NoticeType: noticeTypeApplicationAccepted,
SenderUserID: application.Target.UserID,
TargetUserID: application.Requester.UserID,
PayloadJSON: mustJSON(requesterPayload),
CreatedAtMS: message.OccurredAtMS,
},
{
AppCode: appCode,
EventID: message.EventID,
EventType: message.EventType,
Channel: channelTencentIMC2CAcceptor,
NoticeType: noticeTypeApplicationAccepted,
SenderUserID: application.Requester.UserID,
TargetUserID: application.Target.UserID,
PayloadJSON: mustJSON(acceptorPayload),
CreatedAtMS: message.OccurredAtMS,
},
}
if application.RoomID == "" {
return deliveries
}
roomPayload := cloneJSONMap(payload)
roomPayload["recipient_user_id"] = strconv.FormatInt(application.Requester.UserID, 10)
deliveries = append(deliveries, CPDelivery{
AppCode: appCode,
EventID: message.EventID,
EventType: message.EventType,
Channel: channelTencentIMRoom,
NoticeType: noticeTypeApplicationAccepted,
TargetUserID: application.Requester.UserID,
GroupID: application.RoomID,
PayloadJSON: mustJSON(roomPayload),
CreatedAtMS: message.OccurredAtMS,
})
return deliveries
}
func relationshipCreatedDelivery(appCode string, message usermq.UserOutboxMessage, application cpApplication, relationshipID string, groupID string) ([]CPDelivery, error) {
payload, err := cpRelationshipBroadcastPayload(message, application, relationshipID, groupID)
if err != nil {
return nil, err
}
return []CPDelivery{{
AppCode: appCode,
EventID: message.EventID,
EventType: message.EventType,
Channel: channelTencentIMRegion,
NoticeType: noticeTypeRelationshipCreated,
TargetUserID: 0,
GroupID: groupID,
PayloadJSON: mustJSON(payload),
CreatedAtMS: message.OccurredAtMS,
}}, nil
}
func cpApplicationPayload(message usermq.UserOutboxMessage, application cpApplication, noticeType string, status string) (map[string]any, error) {
if application.ApplicationID == "" || application.Requester.UserID <= 0 || application.Target.UserID <= 0 {
return nil, fmt.Errorf("cp application payload is incomplete")
}
return map[string]any{
"event_id": message.EventID,
"event_type": noticeType,
"source_event_type": message.EventType,
"app_code": appcode.Normalize(message.AppCode),
"application_id": application.ApplicationID,
"relation_type": application.RelationType,
"status": status,
"requester": userPayload(application.Requester),
"target": userPayload(application.Target),
"room_id": application.RoomID,
"room_region_id": application.RoomRegionID,
"gift": giftPayload(application.Gift),
"expires_at_ms": application.ExpiresAtMS,
"decided_at_ms": application.DecidedAtMS,
"source_created_at_ms": message.OccurredAtMS,
}, nil
}
func cpRelationshipBroadcastPayload(message usermq.UserOutboxMessage, application cpApplication, relationshipID string, groupID string) (map[string]any, error) {
if application.Requester.UserID <= 0 || application.Target.UserID <= 0 || application.RoomRegionID <= 0 {
return nil, fmt.Errorf("cp relationship broadcast payload is incomplete")
}
return map[string]any{
"event_id": message.EventID,
"event_type": noticeTypeRelationshipCreated,
"source_event_type": message.EventType,
"app_code": appcode.Normalize(message.AppCode),
"relationship_id": relationshipID,
"application_id": application.ApplicationID,
"relation_type": application.RelationType,
"requester": userPayload(application.Requester),
"target": userPayload(application.Target),
"gift": giftPayload(application.Gift),
"room_id": application.RoomID,
"room_region_id": application.RoomRegionID,
"group_id": groupID,
"source_created_at_ms": message.OccurredAtMS,
}, nil
}
func userPayload(user cpUser) map[string]any {
return map[string]any{
"user_id": strconv.FormatInt(user.UserID, 10),
"display_user_id": user.DisplayUserID,
"username": user.Username,
"avatar": user.Avatar,
}
}
func giftPayload(gift cpGift) map[string]any {
return map[string]any{
"gift_id": gift.GiftID,
"gift_name": gift.GiftName,
"gift_icon_url": gift.GiftIconURL,
"gift_animation_url": gift.GiftAnimationURL,
"gift_count": gift.GiftCount,
"gift_value": gift.GiftValue,
"billing_receipt_id": gift.BillingReceiptID,
}
}
func mustJSON(payload map[string]any) json.RawMessage {
body, _ := json.Marshal(payload)
return body
}
func cloneJSONMap(input map[string]any) map[string]any {
out := make(map[string]any, len(input))
for key, value := range input {
out[key] = value
}
return out
}