diff --git a/cmd/api/main.go b/cmd/api/main.go index 230b96b..95d8a08 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -15,6 +15,7 @@ import ( "chatapp3-golang/internal/service/apppopup" "chatapp3-golang/internal/service/baishun" "chatapp3-golang/internal/service/binancerecharge" + "chatapp3-golang/internal/service/cprelationbroadcast" "chatapp3-golang/internal/service/errorlog" "chatapp3-golang/internal/service/firstrechargereward" "chatapp3-golang/internal/service/gameopen" @@ -71,6 +72,7 @@ func main() { regionIMGroupService := regionimgroup.NewService(app.Config, app.Repository.DB, &app.Gateways) baishunService.SetHighWinBroadcaster(regionIMGroupService) luckyGiftService.SetHighWinBroadcaster(regionIMGroupService) + cpRelationBroadcastService := cprelationbroadcast.NewService(app.Config, app.Repository.DB, regionIMGroupService) voiceRoomRedPacketService := voiceroomredpacket.NewService(app.Config, app.Repository.DB, app.Repository.Redis, app.Gateways.Wallet, regionIMGroupService, app.Gateways.Room) voiceRoomRedPacketService.SetRegionResolver(&app.Gateways) voiceRoomRocketService := voiceroomrocket.NewService(app.Config, app.Repository.DB, app.Repository.Redis, &app.Gateways) @@ -99,6 +101,9 @@ func main() { if err := firstRechargeRewardService.StartMessageConsumer(workerCtx); err != nil { log.Fatalf("start first recharge reward message consumer failed: %v", err) } + if err := cpRelationBroadcastService.StartMessageConsumer(workerCtx); err != nil { + log.Fatalf("start cp relation broadcast message consumer failed: %v", err) + } if err := voiceRoomRocketService.Start(workerCtx); err != nil { log.Fatalf("start voice room rocket workers failed: %v", err) } diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index df1ae7b..a3723c9 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -2,6 +2,7 @@ package main import ( "chatapp3-golang/internal/bootstrap" + "chatapp3-golang/internal/service/cprelationbroadcast" "chatapp3-golang/internal/service/firstrechargereward" "chatapp3-golang/internal/service/regionimgroup" "chatapp3-golang/internal/service/registerreward" @@ -30,6 +31,7 @@ func main() { taskCenterService := taskcenter.NewService(app.Config, app.Repository.DB, &app.Gateways) firstRechargeRewardService := firstrechargereward.NewService(app.Config, app.Repository.DB, &app.Gateways) regionIMGroupService := regionimgroup.NewService(app.Config, app.Repository.DB, &app.Gateways) + cpRelationBroadcastService := cprelationbroadcast.NewService(app.Config, app.Repository.DB, regionIMGroupService) voiceRoomRedPacketService := voiceroomredpacket.NewService(app.Config, app.Repository.DB, app.Repository.Redis, app.Gateways.Wallet, regionIMGroupService, app.Gateways.Room) voiceRoomRedPacketService.SetRegionResolver(&app.Gateways) voiceRoomRocketService := voiceroomrocket.NewService(app.Config, app.Repository.DB, app.Repository.Redis, &app.Gateways) @@ -50,6 +52,9 @@ func main() { if err := firstRechargeRewardService.StartMessageConsumer(workerCtx); err != nil { log.Fatalf("start first recharge reward message consumer failed: %v", err) } + if err := cpRelationBroadcastService.StartMessageConsumer(workerCtx); err != nil { + log.Fatalf("start cp relation broadcast message consumer failed: %v", err) + } if err := voiceRoomRocketService.Start(workerCtx); err != nil { log.Fatalf("start voice room rocket workers failed: %v", err) } diff --git a/internal/config/config.go b/internal/config/config.go index acea750..93e337c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -22,6 +22,7 @@ type Config struct { RoomTurnoverReward RoomTurnoverRewardConfig VoiceRoomRedPacket VoiceRoomRedPacketConfig VoiceRoomRocket VoiceRoomRocketConfig + CPRelationBroadcast CPRelationBroadcastConfig TencentIM TencentIMConfig Baishun BaishunConfig Lingxian LingxianConfig @@ -212,6 +213,25 @@ type VoiceRoomRocketMQConfig struct { Tag string } +// CPRelationBroadcastConfig 保存 CP/兄弟/姐妹关系区域飘屏配置。 +type CPRelationBroadcastConfig struct { + DefaultSysOrigin string + MQ CPRelationBroadcastMQConfig +} + +// CPRelationBroadcastMQConfig 保存 CP 关系成功 MQ 消费配置。 +type CPRelationBroadcastMQConfig struct { + Enabled bool + Endpoint string + Namespace string + AccessKey string + AccessSecret string + SecurityToken string + ConsumerGroup string + Topic string + Tag string +} + // TencentIMConfig 保存腾讯 IM REST API 配置。 type TencentIMConfig struct { AppID int64 @@ -291,6 +311,9 @@ func Load() Config { voiceRoomRocketMQEndpoint := getEnvAny([]string{"CHATAPP_VOICE_ROOM_ROCKET_MQ_ENDPOINT", "LIKEI_ROCKETMQ_ENDPOINT"}, "") voiceRoomRocketMQAccessKey := getEnvAny([]string{"CHATAPP_VOICE_ROOM_ROCKET_MQ_ACCESS_KEY", "LIKEI_ROCKETMQ_ACCESS_KEY"}, "") voiceRoomRocketMQAccessSecret := getEnvAny([]string{"CHATAPP_VOICE_ROOM_ROCKET_MQ_ACCESS_SECRET", "LIKEI_ROCKETMQ_SECRET_KEY"}, "") + cpRelationBroadcastMQEndpoint := getEnvAny([]string{"CHATAPP_CP_RELATION_BROADCAST_MQ_ENDPOINT", "CP_RELATION_BROADCAST_MQ_ENDPOINT", "LIKEI_ROCKETMQ_ENDPOINT"}, "") + cpRelationBroadcastMQAccessKey := getEnvAny([]string{"CHATAPP_CP_RELATION_BROADCAST_MQ_ACCESS_KEY", "CP_RELATION_BROADCAST_MQ_ACCESS_KEY", "LIKEI_ROCKETMQ_ACCESS_KEY"}, "") + cpRelationBroadcastMQAccessSecret := getEnvAny([]string{"CHATAPP_CP_RELATION_BROADCAST_MQ_ACCESS_SECRET", "CP_RELATION_BROADCAST_MQ_SECRET_KEY", "LIKEI_ROCKETMQ_SECRET_KEY"}, "") return Config{ HTTP: HTTPConfig{ @@ -526,6 +549,23 @@ func Load() Config { Tag: getEnvAny([]string{"CHATAPP_VOICE_ROOM_ROCKET_MQ_TAG", "VOICE_ROOM_ROCKET_MQ_TAG"}, "give_gift_v3"), }, }, + CPRelationBroadcast: CPRelationBroadcastConfig{ + DefaultSysOrigin: strings.ToUpper(getEnvAny( + []string{"CHATAPP_CP_RELATION_BROADCAST_DEFAULT_SYS_ORIGIN", "CP_RELATION_BROADCAST_DEFAULT_SYS_ORIGIN", "CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN", "WEEK_STAR_DEFAULT_SYS_ORIGIN"}, + "LIKEI", + )), + MQ: CPRelationBroadcastMQConfig{ + Enabled: getEnvBoolAny([]string{"CHATAPP_CP_RELATION_BROADCAST_MQ_ENABLED", "CP_RELATION_BROADCAST_MQ_ENABLED"}, false), + Endpoint: cpRelationBroadcastMQEndpoint, + Namespace: getEnvAny([]string{"CHATAPP_CP_RELATION_BROADCAST_MQ_NAMESPACE", "CP_RELATION_BROADCAST_MQ_NAMESPACE"}, ""), + AccessKey: cpRelationBroadcastMQAccessKey, + AccessSecret: cpRelationBroadcastMQAccessSecret, + SecurityToken: getEnvAny([]string{"CHATAPP_CP_RELATION_BROADCAST_MQ_SECURITY_TOKEN", "CP_RELATION_BROADCAST_MQ_SECURITY_TOKEN"}, ""), + ConsumerGroup: getEnvAny([]string{"CHATAPP_CP_RELATION_BROADCAST_MQ_CONSUMER_GROUP", "CP_RELATION_BROADCAST_MQ_CONSUMER_GROUP"}, "cp-relation-broadcast"), + Topic: getEnvAny([]string{"CHATAPP_CP_RELATION_BROADCAST_MQ_TOPIC", "CP_RELATION_BROADCAST_MQ_TOPIC"}, "RC_DEFAULT_APP_ORDINARY"), + Tag: getEnvAny([]string{"CHATAPP_CP_RELATION_BROADCAST_MQ_TAG", "CP_RELATION_BROADCAST_MQ_TAG"}, "cp_relation_broadcast"), + }, + }, TencentIM: TencentIMConfig{ AppID: getEnvInt64Any( []string{"CHATAPP_TENCENT_IM_APP_ID", "TENCENT_IM_APP_ID", "LIKEI_IM_APP_ID"}, diff --git a/internal/config/config_test.go b/internal/config/config_test.go index a536ad5..d3dcb1f 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -102,6 +102,27 @@ func TestLoadTaskCenterMQUsesSharedRocketMQCredentials(t *testing.T) { } } +func TestLoadCPRelationBroadcastMQUsesSharedRocketMQCredentials(t *testing.T) { + t.Setenv("CHATAPP_CP_RELATION_BROADCAST_MQ_ENABLED", "true") + t.Setenv("LIKEI_ROCKETMQ_ENDPOINT", "rmq.example.com:8081") + t.Setenv("LIKEI_ROCKETMQ_ACCESS_KEY", "access-key") + t.Setenv("LIKEI_ROCKETMQ_SECRET_KEY", "access-secret") + + cfg := Load() + if !cfg.CPRelationBroadcast.MQ.Enabled { + t.Fatalf("expected cp relation broadcast mq enabled") + } + if got := cfg.CPRelationBroadcast.MQ.Endpoint; got != "rmq.example.com:8081" { + t.Fatalf("cp relation broadcast mq endpoint = %q", got) + } + if got := cfg.CPRelationBroadcast.MQ.Topic; got != "RC_DEFAULT_APP_ORDINARY" { + t.Fatalf("cp relation broadcast mq topic = %q", got) + } + if got := cfg.CPRelationBroadcast.MQ.Tag; got != "cp_relation_broadcast" { + t.Fatalf("cp relation broadcast mq tag = %q", got) + } +} + func TestLoadTaskCenterArchiveUsesSharedCOSCredentials(t *testing.T) { t.Setenv("CHATAPP_TASK_CENTER_ARCHIVE_ENABLED", "true") t.Setenv("LIKEI_TENCENT_COS_BUCKET", "bucket-123") diff --git a/internal/router/manager_center_routes.go b/internal/router/manager_center_routes.go index b38408d..d02059f 100644 --- a/internal/router/manager_center_routes.go +++ b/internal/router/manager_center_routes.go @@ -60,6 +60,29 @@ func registerManagerCenterRoutes(engine *gin.Engine, javaClient authGateway, ser writeOK(c, resp) }) + group.GET("/bd-leaders", func(c *gin.Context) { + resp, err := service.ListBDLeaders(c.Request.Context(), mustAuthUser(c)) + if err != nil { + writeError(c, err) + return + } + writeOK(c, resp) + }) + + group.POST("/bd-leaders", func(c *gin.Context) { + var req managercenter.AddBDLeaderRequest + if err := c.ShouldBindJSON(&req); err != nil { + writeError(c, managercenter.NewAppError(http.StatusBadRequest, "bad_request", err.Error())) + return + } + resp, err := service.AddBDLeader(c.Request.Context(), mustAuthUser(c), req) + if err != nil { + writeError(c, err) + return + } + writeOK(c, resp) + }) + group.POST("/users/ban", func(c *gin.Context) { var req managercenter.BanUserRequest if err := c.ShouldBindJSON(&req); err != nil { diff --git a/internal/service/cprelationbroadcast/event.go b/internal/service/cprelationbroadcast/event.go new file mode 100644 index 0000000..ce45a90 --- /dev/null +++ b/internal/service/cprelationbroadcast/event.go @@ -0,0 +1,327 @@ +package cprelationbroadcast + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + + "chatapp3-golang/internal/common" + "chatapp3-golang/internal/model" + "chatapp3-golang/internal/service/regionimgroup" + + "gorm.io/gorm" +) + +// ProcessPayload 解码 Java MQ 消息并发送区域飘屏。 +func (s *Service) ProcessPayload(ctx context.Context, payload string) (*regionimgroup.RegionBroadcastResponse, error) { + req, err := decodeEventPayload(payload, s.cfg.CPRelationBroadcast.MQ.Tag) + if err != nil { + return nil, err + } + if req.isEmpty() { + return nil, nil + } + return s.ProcessEvent(ctx, req) +} + +// ProcessEvent 处理标准化后的 CP 关系事件。 +func (s *Service) ProcessEvent(ctx context.Context, req EventRequest) (*regionimgroup.RegionBroadcastResponse, error) { + if s == nil || s.broadcaster == nil { + return nil, common.NewAppError(http.StatusInternalServerError, "cp_relation_broadcaster_missing", "cp relation broadcaster is missing") + } + if s.db == nil { + return nil, common.NewAppError(http.StatusInternalServerError, "cp_relation_db_missing", "cp relation db is missing") + } + req = normalizeEventRequest(req) + + userID := req.UserID.Int64() + cpUserID := req.CpUserID.Int64() + if userID <= 0 || cpUserID <= 0 { + return nil, common.NewAppError(http.StatusBadRequest, "invalid_cp_relation_users", "userId and cpUserId are required") + } + relationType, err := normalizeRelationType(req.RelationType) + if err != nil { + return nil, err + } + sysOrigin := normalizeSysOrigin(req.SysOrigin, s.cfg.CPRelationBroadcast.DefaultSysOrigin) + + user, err := s.loadEventUser(ctx, userID, req.User) + if err != nil { + return nil, err + } + cpUser, err := s.loadEventUser(ctx, cpUserID, req.CpUser) + if err != nil { + return nil, err + } + + msg := relationMessage(relationType, user.displayName(), cpUser.displayName()) + data := buildBroadcastData(req, sysOrigin, relationType, user, cpUser, msg) + return s.broadcaster.SendRegionBroadcast(ctx, regionimgroup.RegionBroadcastRequest{ + SysOrigin: sysOrigin, + Type: MessageTypeCPRelationBroadcast, + Data: data, + }) +} + +func (s *Service) loadEventUser(ctx context.Context, userID int64, provided *EventUser) (EventUser, error) { + var user model.UserBaseInfo + err := s.db.WithContext(ctx).Where("id = ?", userID).First(&user).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + account := strconv.FormatInt(userID, 10) + user = model.UserBaseInfo{ + ID: userID, + Account: account, + UserNickname: account, + } + } else if err != nil { + return EventUser{}, err + } + if strings.TrimSpace(user.Account) == "" { + user.Account = strconv.FormatInt(userID, 10) + } + if strings.TrimSpace(user.UserNickname) == "" { + user.UserNickname = user.Account + } + return mergeEventUser(userID, user, provided), nil +} + +func buildBroadcastData(req EventRequest, sysOrigin, relationType string, user EventUser, cpUser EventUser, msg string) map[string]any { + userID := user.normalizedID() + cpUserID := cpUser.normalizedID() + userIDText := strconv.FormatInt(userID, 10) + cpUserIDText := strconv.FormatInt(cpUserID, 10) + + data := map[string]any{ + "sysOrigin": sysOrigin, + "type": MessageTypeCPRelationBroadcast, + "relationType": relationType, + "relationLabel": relationLabel(relationType), + "msg": msg, + "userId": userIDText, + "sendUserId": userIDText, + "senderUserId": userIDText, + "cpUserId": cpUserIDText, + "acceptUserId": cpUserIDText, + "toUserId": cpUserIDText, + "user": user.toMap(), + "cpUser": cpUser.toMap(), + + "account": user.Account, + "actualAccount": user.Account, + "userNickname": user.displayName(), + "nickname": user.displayName(), + "userAvatar": firstNonBlank(user.UserAvatar, user.Avatar), + "countryCode": user.CountryCode, + "countryName": user.CountryName, + "cpUserAccount": cpUser.Account, + "cpUserNickname": cpUser.displayName(), + "cpUserAvatar": firstNonBlank(cpUser.UserAvatar, cpUser.Avatar), + "cpUserCountryCode": cpUser.CountryCode, + "cpUserCountryName": cpUser.CountryName, + "acceptAccount": cpUser.Account, + "acceptNickname": cpUser.displayName(), + "acceptUserAvatar": firstNonBlank(cpUser.UserAvatar, cpUser.Avatar), + "toUserName": cpUser.displayName(), + "toUserAvatarUrl": firstNonBlank(cpUser.UserAvatar, cpUser.Avatar), + } + if eventID := strings.TrimSpace(req.EventID); eventID != "" { + data["eventId"] = eventID + } + if applyID := req.ApplyID.Int64(); applyID > 0 { + data["applyId"] = strconv.FormatInt(applyID, 10) + } + if occurredAt := strings.TrimSpace(req.OccurredAt); occurredAt != "" { + data["occurredAt"] = occurredAt + } + if len(req.Payload) > 0 { + data["payload"] = req.Payload + } + return data +} + +func normalizeEventRequest(req EventRequest) EventRequest { + if req.UserID.Int64() <= 0 { + req.UserID = req.SendUserID + } + if req.CpUserID.Int64() <= 0 { + req.CpUserID = req.AcceptUserID + } + if req.UserID.Int64() <= 0 && req.User != nil { + req.UserID = flexibleInt64(req.User.normalizedID()) + } + if req.CpUserID.Int64() <= 0 && req.CpUser != nil { + req.CpUserID = flexibleInt64(req.CpUser.normalizedID()) + } + return req +} + +func (req EventRequest) isEmpty() bool { + return strings.TrimSpace(req.EventID) == "" && + strings.TrimSpace(req.RelationType) == "" && + req.UserID.Int64() <= 0 && + req.CpUserID.Int64() <= 0 && + req.SendUserID.Int64() <= 0 && + req.AcceptUserID.Int64() <= 0 +} + +func normalizeRelationType(value string) (string, error) { + normalized := strings.ToUpper(strings.TrimSpace(value)) + if normalized == "" { + return RelationTypeCP, nil + } + switch normalized { + case RelationTypeCP: + return RelationTypeCP, nil + case RelationTypeBrother, "BROTHERS": + return RelationTypeBrother, nil + case RelationTypeSisters, "SISTER": + return RelationTypeSisters, nil + default: + return "", common.NewAppError(http.StatusBadRequest, "invalid_cp_relation_type", fmt.Sprintf("invalid cp relationType: %s", value)) + } +} + +func relationLabel(relationType string) string { + switch relationType { + case RelationTypeBrother: + return "brothers" + case RelationTypeSisters: + return "sisters" + default: + return "cp_relation" + } +} + +func relationMessage(relationType, userName, cpUserName string) string { + switch relationType { + case RelationTypeBrother: + return fmt.Sprintf("%s与%s结为兄弟", userName, cpUserName) + case RelationTypeSisters: + return fmt.Sprintf("%s与%s结为姐妹", userName, cpUserName) + default: + return fmt.Sprintf("%s与%s get cp relation", userName, cpUserName) + } +} + +func normalizeSysOrigin(values ...string) string { + for _, value := range values { + if text := strings.ToUpper(strings.TrimSpace(value)); text != "" { + return text + } + } + return defaultSysOrigin +} + +func decodeEventPayload(payload string, expectedTag string) (EventRequest, error) { + payload = strings.TrimSpace(payload) + if payload == "" { + return EventRequest{}, nil + } + req, err := decodeEventObject([]byte(payload)) + if err != nil { + return EventRequest{}, err + } + if !req.isEmpty() { + return req, nil + } + + var envelope messageEventEnvelope + if err := json.Unmarshal([]byte(payload), &envelope); err != nil { + return EventRequest{}, err + } + if shouldSkipEnvelope(envelope.Tag, expectedTag) { + return EventRequest{}, nil + } + bodyPayload, ok, err := envelope.bodyPayload() + if err != nil || !ok { + return EventRequest{}, err + } + return decodeEventObject([]byte(bodyPayload)) +} + +type rawEventRequest struct { + EventRequest + SysOriginSnake string `json:"sys_origin"` + EventIDSnake string `json:"event_id"` + ApplyIDSnake flexibleInt64 `json:"apply_id"` + RelationTypeSnake string `json:"relation_type"` + UserIDSnake flexibleInt64 `json:"user_id"` + CpUserIDSnake flexibleInt64 `json:"cp_user_id"` + SendUserIDSnake flexibleInt64 `json:"send_user_id"` + AcceptUserIDSnake flexibleInt64 `json:"accept_user_id"` + OccurredAtSnake string `json:"occurred_at"` + TargetUserID flexibleInt64 `json:"targetUserId"` + TargetUserIDSnake flexibleInt64 `json:"target_user_id"` + ToUserIDSnake flexibleInt64 `json:"to_user_id"` +} + +func decodeEventObject(data []byte) (EventRequest, error) { + var raw rawEventRequest + if err := json.Unmarshal(data, &raw); err != nil { + return EventRequest{}, err + } + req := raw.EventRequest + req.SysOrigin = firstNonBlank(req.SysOrigin, raw.SysOriginSnake) + req.EventID = firstNonBlank(req.EventID, raw.EventIDSnake) + req.RelationType = firstNonBlank(req.RelationType, raw.RelationTypeSnake) + req.OccurredAt = firstNonBlank(req.OccurredAt, raw.OccurredAtSnake) + if req.ApplyID.Int64() <= 0 { + req.ApplyID = raw.ApplyIDSnake + } + if req.UserID.Int64() <= 0 { + req.UserID = firstFlexible(raw.UserIDSnake, req.SendUserID, raw.SendUserIDSnake) + } + if req.CpUserID.Int64() <= 0 { + req.CpUserID = firstFlexible(raw.CpUserIDSnake, req.AcceptUserID, raw.AcceptUserIDSnake, raw.TargetUserID, raw.TargetUserIDSnake, raw.ToUserIDSnake) + } + if req.SendUserID.Int64() <= 0 { + req.SendUserID = firstFlexible(req.UserID, raw.SendUserIDSnake) + } + if req.AcceptUserID.Int64() <= 0 { + req.AcceptUserID = firstFlexible(req.CpUserID, raw.AcceptUserIDSnake, raw.TargetUserID, raw.TargetUserIDSnake, raw.ToUserIDSnake) + } + return req, nil +} + +func firstFlexible(values ...flexibleInt64) flexibleInt64 { + for _, value := range values { + if value.Int64() > 0 { + return value + } + } + return 0 +} + +type messageEventEnvelope struct { + Tag string `json:"tag"` + Body json.RawMessage `json:"body"` +} + +func (e messageEventEnvelope) bodyPayload() (string, bool, error) { + raw := strings.TrimSpace(string(e.Body)) + if raw == "" || raw == "null" { + return "", false, nil + } + if strings.HasPrefix(raw, "\"") { + var body string + if err := json.Unmarshal(e.Body, &body); err != nil { + return "", false, err + } + body = strings.TrimSpace(body) + return body, body != "", nil + } + return raw, true, nil +} + +func shouldSkipEnvelope(actualTag, expectedTag string) bool { + actualTag = strings.TrimSpace(actualTag) + expectedTag = strings.TrimSpace(expectedTag) + if actualTag == "" || expectedTag == "" || expectedTag == "*" { + return false + } + return actualTag != expectedTag +} diff --git a/internal/service/cprelationbroadcast/event_test.go b/internal/service/cprelationbroadcast/event_test.go new file mode 100644 index 0000000..d449a49 --- /dev/null +++ b/internal/service/cprelationbroadcast/event_test.go @@ -0,0 +1,142 @@ +package cprelationbroadcast + +import ( + "context" + "testing" + "time" + + "chatapp3-golang/internal/config" + "chatapp3-golang/internal/model" + "chatapp3-golang/internal/service/regionimgroup" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func TestProcessEventBuildsCPRelationBroadcastPayload(t *testing.T) { + service, broadcaster := newCPRelationBroadcastTestService(t) + + _, err := service.ProcessEvent(context.Background(), EventRequest{ + SysOrigin: "LIKEI", + EventID: "CP_RELATION:9001", + ApplyID: flexibleInt64(9001), + RelationType: RelationTypeCP, + UserID: flexibleInt64(1001), + CpUserID: flexibleInt64(1002), + OccurredAt: "2026-05-22T12:00:00Z", + }) + if err != nil { + t.Fatalf("ProcessEvent() error = %v", err) + } + if len(broadcaster.requests) != 1 { + t.Fatalf("requests = %d, want 1", len(broadcaster.requests)) + } + req := broadcaster.requests[0] + if req.Type != MessageTypeCPRelationBroadcast { + t.Fatalf("type = %q", req.Type) + } + data := req.Data + if data["msg"] != "Alice与Bob get cp relation" || + data["relationType"] != RelationTypeCP || + data["userId"] != "1001" || + data["cpUserId"] != "1002" { + t.Fatalf("payload = %+v", data) + } + user, ok := data["user"].(map[string]any) + if !ok || user["userNickname"] != "Alice" || user["account"] != "alice" { + t.Fatalf("user payload = %+v", data["user"]) + } + cpUser, ok := data["cpUser"].(map[string]any) + if !ok || cpUser["userNickname"] != "Bob" || cpUser["account"] != "bob" { + t.Fatalf("cpUser payload = %+v", data["cpUser"]) + } +} + +func TestProcessEventBuildsBrotherAndSistersMessages(t *testing.T) { + service, broadcaster := newCPRelationBroadcastTestService(t) + cases := []struct { + relationType string + wantMsg string + }{ + {RelationTypeBrother, "Alice与Bob结为兄弟"}, + {RelationTypeSisters, "Alice与Bob结为姐妹"}, + } + + for _, tc := range cases { + broadcaster.requests = nil + _, err := service.ProcessEvent(context.Background(), EventRequest{ + RelationType: tc.relationType, + UserID: flexibleInt64(1001), + CpUserID: flexibleInt64(1002), + }) + if err != nil { + t.Fatalf("ProcessEvent(%s) error = %v", tc.relationType, err) + } + if len(broadcaster.requests) != 1 { + t.Fatalf("requests = %d, want 1", len(broadcaster.requests)) + } + if got := broadcaster.requests[0].Data["msg"]; got != tc.wantMsg { + t.Fatalf("msg = %v, want %s", got, tc.wantMsg) + } + } +} + +func TestProcessPayloadDecodesMessageEventEnvelopeAndAliases(t *testing.T) { + service, broadcaster := newCPRelationBroadcastTestService(t) + service.cfg.CPRelationBroadcast.MQ.Tag = "cp_relation_broadcast" + payload := `{"tag":"cp_relation_broadcast","body":"{\"sys_origin\":\"likei\",\"event_id\":\"CP_RELATION:9002\",\"apply_id\":\"9002\",\"relation_type\":\"BROTHERS\",\"send_user_id\":\"1001\",\"accept_user_id\":\"1002\"}"}` + + _, err := service.ProcessPayload(context.Background(), payload) + if err != nil { + t.Fatalf("ProcessPayload() error = %v", err) + } + if len(broadcaster.requests) != 1 { + t.Fatalf("requests = %d, want 1", len(broadcaster.requests)) + } + req := broadcaster.requests[0] + if req.SysOrigin != "LIKEI" { + t.Fatalf("sysOrigin = %q", req.SysOrigin) + } + if req.Data["eventId"] != "CP_RELATION:9002" || + req.Data["applyId"] != "9002" || + req.Data["relationType"] != RelationTypeBrother || + req.Data["msg"] != "Alice与Bob结为兄弟" { + t.Fatalf("payload = %+v", req.Data) + } +} + +func newCPRelationBroadcastTestService(t *testing.T) (*Service, *fakeCPRegionBroadcaster) { + t.Helper() + db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + if err := db.AutoMigrate(&model.UserBaseInfo{}); err != nil { + t.Fatalf("auto migrate: %v", err) + } + now := time.Now() + if err := db.Create(&[]model.UserBaseInfo{ + {ID: 1001, Account: "alice", UserNickname: "Alice", UserAvatar: "alice.png", OriginSys: "LIKEI", CountryCode: "SA", CountryName: "Saudi Arabia", CreateTime: now}, + {ID: 1002, Account: "bob", UserNickname: "Bob", UserAvatar: "bob.png", OriginSys: "LIKEI", CountryCode: "SA", CountryName: "Saudi Arabia", CreateTime: now}, + }).Error; err != nil { + t.Fatalf("seed users: %v", err) + } + broadcaster := &fakeCPRegionBroadcaster{} + return NewService(config.Config{ + CPRelationBroadcast: config.CPRelationBroadcastConfig{DefaultSysOrigin: "LIKEI"}, + }, db, broadcaster), broadcaster +} + +type fakeCPRegionBroadcaster struct { + requests []regionimgroup.RegionBroadcastRequest +} + +func (f *fakeCPRegionBroadcaster) SendRegionBroadcast(_ context.Context, req regionimgroup.RegionBroadcastRequest) (*regionimgroup.RegionBroadcastResponse, error) { + f.requests = append(f.requests, req) + return ®ionimgroup.RegionBroadcastResponse{ + SysOrigin: req.SysOrigin, + RegionCode: "AR", + GroupID: "@region-ar", + Type: req.Type, + }, nil +} diff --git a/internal/service/cprelationbroadcast/lifecycle.go b/internal/service/cprelationbroadcast/lifecycle.go new file mode 100644 index 0000000..9811cca --- /dev/null +++ b/internal/service/cprelationbroadcast/lifecycle.go @@ -0,0 +1,139 @@ +package cprelationbroadcast + +import ( + "context" + "encoding/json" + "errors" + "log" + "net/http" + "strings" + "time" + + "chatapp3-golang/internal/common" + + rmq "github.com/apache/rocketmq-clients/golang/v5" + "github.com/apache/rocketmq-clients/golang/v5/credentials" +) + +// StartMessageConsumer 启动 CP 关系成功 MQ 消费链路。 +func (s *Service) StartMessageConsumer(ctx context.Context) error { + if !s.cfg.CPRelationBroadcast.MQ.Enabled { + log.Printf("cp relation broadcast rocketmq consumer disabled") + return nil + } + if strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.Endpoint) == "" || + strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.AccessKey) == "" || + strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.AccessSecret) == "" || + strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.Topic) == "" || + strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.ConsumerGroup) == "" { + log.Printf("cp relation broadcast rocketmq consumer skipped: endpoint, credentials, topic or consumer group missing") + return nil + } + + go s.startRocketMQConsumerWithRetry(ctx) + return nil +} + +func (s *Service) startRocketMQConsumerWithRetry(ctx context.Context) { + retryDelay := 5 * time.Second + for { + if ctx.Err() != nil { + return + } + if err := s.startRocketMQConsumer(ctx); err != nil { + log.Printf("start cp relation broadcast rocketmq consumer failed: %v; retry in %s", err, retryDelay) + timer := time.NewTimer(retryDelay) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + if retryDelay < time.Minute { + retryDelay *= 2 + if retryDelay > time.Minute { + retryDelay = time.Minute + } + } + continue + } + return + } +} + +func (s *Service) startRocketMQConsumer(ctx context.Context) error { + filter := rmq.SUB_ALL + if tag := strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.Tag); tag != "" && tag != "*" { + filter = rmq.NewFilterExpression(tag) + } + + consumer, err := rmq.NewPushConsumer(&rmq.Config{ + Endpoint: s.cfg.CPRelationBroadcast.MQ.Endpoint, + NameSpace: s.cfg.CPRelationBroadcast.MQ.Namespace, + ConsumerGroup: s.cfg.CPRelationBroadcast.MQ.ConsumerGroup, + Credentials: &credentials.SessionCredentials{ + AccessKey: s.cfg.CPRelationBroadcast.MQ.AccessKey, + AccessSecret: s.cfg.CPRelationBroadcast.MQ.AccessSecret, + SecurityToken: s.cfg.CPRelationBroadcast.MQ.SecurityToken, + }, + }, + rmq.WithPushSubscriptionExpressions(map[string]*rmq.FilterExpression{ + s.cfg.CPRelationBroadcast.MQ.Topic: filter, + }), + rmq.WithPushConsumptionThreadCount(4), + rmq.WithPushMaxCacheMessageCount(256), + rmq.WithPushMessageListener(&rmq.FuncMessageListener{ + Consume: func(messageView *rmq.MessageView) rmq.ConsumerResult { + if messageView == nil { + return rmq.SUCCESS + } + if err := s.processRocketMQMessage(context.Background(), messageView); err != nil { + log.Printf("cp relation broadcast mq process failed. messageId=%s err=%v", messageView.GetMessageId(), err) + return rmq.FAILURE + } + return rmq.SUCCESS + }, + }), + ) + if err != nil { + return err + } + if err := consumer.Start(); err != nil { + return err + } + s.rocketConsumer = consumer + log.Printf("cp relation broadcast rocketmq consumer started. topic=%s group=%s tag=%s", + s.cfg.CPRelationBroadcast.MQ.Topic, + s.cfg.CPRelationBroadcast.MQ.ConsumerGroup, + s.cfg.CPRelationBroadcast.MQ.Tag, + ) + go func() { + <-ctx.Done() + if s.rocketConsumer != nil { + _ = s.rocketConsumer.GracefulStop() + } + }() + return nil +} + +func (s *Service) processRocketMQMessage(ctx context.Context, messageView *rmq.MessageView) error { + payload := strings.TrimSpace(string(messageView.GetBody())) + if payload == "" { + return nil + } + if _, err := s.ProcessPayload(ctx, payload); err != nil { + var syntaxErr *json.SyntaxError + var typeErr *json.UnmarshalTypeError + if errors.As(err, &syntaxErr) || errors.As(err, &typeErr) { + log.Printf("drop malformed cp relation broadcast message. messageId=%s err=%v", messageView.GetMessageId(), err) + return nil + } + var appErr *common.AppError + if errors.As(err, &appErr) && appErr.Status >= http.StatusBadRequest && appErr.Status < http.StatusInternalServerError { + log.Printf("drop invalid cp relation broadcast message. messageId=%s code=%s err=%v", messageView.GetMessageId(), appErr.Code, err) + return nil + } + return err + } + return nil +} diff --git a/internal/service/cprelationbroadcast/types.go b/internal/service/cprelationbroadcast/types.go new file mode 100644 index 0000000..6fbcacf --- /dev/null +++ b/internal/service/cprelationbroadcast/types.go @@ -0,0 +1,199 @@ +package cprelationbroadcast + +import ( + "context" + "fmt" + "strconv" + "strings" + + "chatapp3-golang/internal/config" + "chatapp3-golang/internal/model" + "chatapp3-golang/internal/service/regionimgroup" + + rmq "github.com/apache/rocketmq-clients/golang/v5" + "gorm.io/gorm" +) + +const ( + MessageTypeCPRelationBroadcast = "CP_RELATION_BROADCAST" + + RelationTypeCP = "CP" + RelationTypeBrother = "BROTHER" + RelationTypeSisters = "SISTERS" + + defaultSysOrigin = "LIKEI" +) + +type dbHandle interface { + WithContext(ctx context.Context) *gorm.DB +} + +type regionBroadcaster interface { + SendRegionBroadcast(ctx context.Context, req regionimgroup.RegionBroadcastRequest) (*regionimgroup.RegionBroadcastResponse, error) +} + +// Service 消费 Java CP 关系成功 MQ，并复用区域 IM 广播发送飘屏。 +type Service struct { + cfg config.Config + db dbHandle + broadcaster regionBroadcaster + rocketConsumer rmq.PushConsumer +} + +func NewService(cfg config.Config, db dbHandle, broadcaster regionBroadcaster) *Service { + return &Service{cfg: cfg, db: db, broadcaster: broadcaster} +} + +// EventRequest 是 Java 在 CP/兄弟/姐妹关系达成后投递给 Go 的 MQ 事件。 +type EventRequest struct { + SysOrigin string `json:"sysOrigin"` + EventID string `json:"eventId"` + ApplyID flexibleInt64 `json:"applyId"` + RelationType string `json:"relationType"` + UserID flexibleInt64 `json:"userId"` + CpUserID flexibleInt64 `json:"cpUserId"` + SendUserID flexibleInt64 `json:"sendUserId"` + AcceptUserID flexibleInt64 `json:"acceptUserId"` + OccurredAt string `json:"occurredAt"` + User *EventUser `json:"user"` + CpUser *EventUser `json:"cpUser"` + Payload map[string]any `json:"payload,omitempty"` +} + +type EventUser struct { + ID flexibleInt64 `json:"id"` + UserID flexibleInt64 `json:"userId"` + Account string `json:"account"` + Nickname string `json:"nickname"` + UserNickname string `json:"userNickname"` + Avatar string `json:"avatar"` + UserAvatar string `json:"userAvatar"` + CountryCode string `json:"countryCode"` + CountryName string `json:"countryName"` + OriginSys string `json:"originSys"` +} + +func (u EventUser) normalizedID() int64 { + if id := u.UserID.Int64(); id > 0 { + return id + } + return u.ID.Int64() +} + +func (u EventUser) displayName() string { + for _, value := range []string{u.UserNickname, u.Nickname, u.Account} { + if text := strings.TrimSpace(value); text != "" { + return text + } + } + if id := u.normalizedID(); id > 0 { + return strconv.FormatInt(id, 10) + } + return "" +} + +func (u EventUser) toMap() map[string]any { + id := u.normalizedID() + data := map[string]any{} + if id > 0 { + data["userId"] = strconv.FormatInt(id, 10) + data["id"] = strconv.FormatInt(id, 10) + } + if account := strings.TrimSpace(u.Account); account != "" { + data["account"] = account + } + if nickname := strings.TrimSpace(firstNonBlank(u.UserNickname, u.Nickname)); nickname != "" { + data["userNickname"] = nickname + data["nickname"] = nickname + } + if avatar := strings.TrimSpace(firstNonBlank(u.UserAvatar, u.Avatar)); avatar != "" { + data["userAvatar"] = avatar + data["avatar"] = avatar + } + if countryCode := strings.TrimSpace(u.CountryCode); countryCode != "" { + data["countryCode"] = countryCode + } + if countryName := strings.TrimSpace(u.CountryName); countryName != "" { + data["countryName"] = countryName + } + if originSys := strings.TrimSpace(u.OriginSys); originSys != "" { + data["originSys"] = originSys + } + return data +} + +func eventUserFromModel(user model.UserBaseInfo) EventUser { + return EventUser{ + ID: flexibleInt64(user.ID), + UserID: flexibleInt64(user.ID), + Account: user.Account, + Nickname: user.UserNickname, + UserNickname: user.UserNickname, + Avatar: user.UserAvatar, + UserAvatar: user.UserAvatar, + CountryCode: user.CountryCode, + CountryName: user.CountryName, + OriginSys: user.OriginSys, + } +} + +func mergeEventUser(id int64, dbUser model.UserBaseInfo, provided *EventUser) EventUser { + merged := eventUserFromModel(dbUser) + if id > 0 { + merged.ID = flexibleInt64(id) + merged.UserID = flexibleInt64(id) + } + if provided == nil { + return merged + } + if provided.normalizedID() > 0 { + merged.ID = flexibleInt64(provided.normalizedID()) + merged.UserID = flexibleInt64(provided.normalizedID()) + } + merged.Account = firstNonBlank(provided.Account, merged.Account) + merged.UserNickname = firstNonBlank(provided.UserNickname, provided.Nickname, merged.UserNickname) + merged.Nickname = merged.UserNickname + merged.UserAvatar = firstNonBlank(provided.UserAvatar, provided.Avatar, merged.UserAvatar) + merged.Avatar = merged.UserAvatar + merged.CountryCode = firstNonBlank(provided.CountryCode, merged.CountryCode) + merged.CountryName = firstNonBlank(provided.CountryName, merged.CountryName) + merged.OriginSys = firstNonBlank(provided.OriginSys, merged.OriginSys) + return merged +} + +type flexibleInt64 int64 + +func (v *flexibleInt64) UnmarshalJSON(data []byte) error { + raw := strings.TrimSpace(string(data)) + switch raw { + case "", "null", `""`: + *v = 0 + return nil + } + if strings.HasPrefix(raw, `"`) && strings.HasSuffix(raw, `"`) { + unquoted, err := strconv.Unquote(raw) + if err != nil { + return err + } + raw = strings.TrimSpace(unquoted) + } + parsed, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return fmt.Errorf("invalid int64 value: %w", err) + } + *v = flexibleInt64(parsed) + return nil +} + +func (v flexibleInt64) Int64() int64 { + return int64(v) +} + +func firstNonBlank(values ...string) string { + for _, value := range values { + if text := strings.TrimSpace(value); text != "" { + return text + } + } + return "" +} diff --git a/internal/service/managercenter/bd_leader.go b/internal/service/managercenter/bd_leader.go new file mode 100644 index 0000000..3dcb495 --- /dev/null +++ b/internal/service/managercenter/bd_leader.go @@ -0,0 +1,367 @@ +package managercenter + +import ( + "context" + "errors" + "strings" + "time" + + "chatapp3-golang/internal/utils" + + "gorm.io/gorm" +) + +type bdLeaderRow struct { + ID int64 `gorm:"column:id;primaryKey"` + Origin string `gorm:"column:sys_origin"` + UserID int64 `gorm:"column:user_id"` + Contact string `gorm:"column:contact"` + RegionID string `gorm:"column:region_id"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` + CreateUser int64 `gorm:"column:create_user"` + UpdateUser int64 `gorm:"column:update_user"` +} + +func (bdLeaderRow) TableName() string { + return "room_bd_lead_base_info" +} + +type businessDevelopmentBaseInfoRow struct { + ID int64 `gorm:"column:id;primaryKey"` + Origin string `gorm:"column:sys_origin"` + BDLeadUserID *int64 `gorm:"column:bd_lead_user_id"` + UserID int64 `gorm:"column:user_id"` + Contact string `gorm:"column:contact"` + Region string `gorm:"column:region"` + MemberQuantity int64 `gorm:"column:member_quantity"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` + CreateUser int64 `gorm:"column:create_user"` + UpdateUser int64 `gorm:"column:update_user"` +} + +func (businessDevelopmentBaseInfoRow) TableName() string { + return "room_business_development_base_info" +} + +type userHistoryIdentityRow struct { + UserID int64 `gorm:"column:user_id;primaryKey"` + IsBD bool `gorm:"column:is_bd"` + IsAgent bool `gorm:"column:is_agent"` + IsHost bool `gorm:"column:is_host"` + IsManager bool `gorm:"column:is_manager"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` + CreateUser int64 `gorm:"column:create_user"` + UpdateUser int64 `gorm:"column:update_user"` +} + +func (userHistoryIdentityRow) TableName() string { + return "user_history_identity" +} + +type administratorRow struct { + ID int64 `gorm:"column:id;primaryKey"` + UserID int64 `gorm:"column:user_id"` + Status bool `gorm:"column:status"` +} + +func (administratorRow) TableName() string { + return "sys_administrator" +} + +type bdLeaderListRow struct { + ID int64 `gorm:"column:id"` + UserID int64 `gorm:"column:user_id"` + Origin string `gorm:"column:sys_origin"` + Contact string `gorm:"column:contact"` + RegionID string `gorm:"column:region_id"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` + Account string `gorm:"column:account"` + UserAvatar string `gorm:"column:user_avatar"` + UserNickname string `gorm:"column:user_nickname"` +} + +func (s *Service) ListBDLeaders(ctx context.Context, user AuthUser) (*BDLeaderListResponse, error) { + manager, err := s.requireManagerInfo(ctx, user) + if err != nil { + return nil, err + } + + query := s.db.WithContext(ctx). + Table("room_bd_lead_base_info AS leader"). + Select(strings.Join([]string{ + "leader.id", + "leader.user_id", + "leader.sys_origin", + "leader.contact", + "leader.region_id", + "leader.create_time", + "leader.update_time", + "user_info.account", + "user_info.user_avatar", + "user_info.user_nickname", + }, ", ")). + Joins("LEFT JOIN user_base_info AS user_info ON user_info.id = leader.user_id AND user_info.is_del = ?", false). + Where("leader.create_user = ?", user.UserID) + + if origin := managerSysOrigin(manager, user); origin != "" { + query = query.Where("leader.sys_origin = ?", origin) + } + + var rows []bdLeaderListRow + if err := query.Order("leader.create_time DESC, leader.id DESC").Find(&rows).Error; err != nil { + return nil, serverError("bd_leader_query_failed", err.Error()) + } + + records := make([]BDLeaderView, 0, len(rows)) + for _, row := range rows { + records = append(records, bdLeaderListRowToView(row)) + } + return &BDLeaderListResponse{Count: len(records), Records: records}, nil +} + +func (s *Service) AddBDLeader(ctx context.Context, user AuthUser, req AddBDLeaderRequest) (*AddBDLeaderResponse, error) { + manager, err := s.requireManagerInfo(ctx, user) + if err != nil { + return nil, err + } + if req.UserID.Int64() <= 0 { + return nil, badRequest("user_required", "userId is required") + } + + target, err := s.loadUserRowByID(ctx, req.UserID.Int64()) + if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, notFound("user_not_found", "user not found") + } + return nil, serverError("user_query_failed", err.Error()) + } + + sysOrigin := managerSysOrigin(manager, user) + if sysOrigin == "" { + sysOrigin = normalizeSysOrigin(target.OriginSys) + } + if target.OriginSys != "" && sysOrigin != "" && !strings.EqualFold(target.OriginSys, sysOrigin) { + return nil, badRequest("bd_leader_origin_mismatch", "target user sysOrigin does not match manager") + } + + regionID := strings.TrimSpace(req.RegionID) + if regionID == "" { + regionID = strings.TrimSpace(manager.RegionID) + } + if regionID == "" { + return nil, badRequest("manager_region_required", "manager region is required") + } + + if exists, err := s.userIsAdministrator(ctx, target.ID); err != nil { + return nil, err + } else if exists { + return nil, badRequest("user_is_admin", "administrator cannot be added as BD leader") + } + + if exists, err := s.userIsBDLeader(ctx, target.ID); err != nil { + return nil, err + } else if exists { + return nil, badRequest("user_already_bd_leader", "user is already a BD leader") + } + + leaderID, err := utils.NextID() + if err != nil { + return nil, serverError("id_generate_failed", err.Error()) + } + now := time.Now() + contact := strings.TrimSpace(req.Contact) + leader := bdLeaderRow{ + ID: leaderID, + Origin: sysOrigin, + UserID: target.ID, + Contact: contact, + RegionID: regionID, + CreateTime: now, + UpdateTime: now, + CreateUser: user.UserID, + UpdateUser: user.UserID, + } + + if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Create(&leader).Error; err != nil { + return serverError("bd_leader_create_failed", err.Error()) + } + if err := upsertBDBaseInfo(tx, leader, user.UserID, now); err != nil { + return err + } + if err := upsertUserHistoryBD(tx, target.ID, user.UserID, now); err != nil { + return err + } + return nil + }); err != nil { + return nil, err + } + + return &AddBDLeaderResponse{ + Success: true, + Leader: bdLeaderRowToView(leader, target), + }, nil +} + +func (s *Service) requireManagerInfo(ctx context.Context, user AuthUser) (teamManagerInfoRow, error) { + if user.UserID <= 0 { + return teamManagerInfoRow{}, unauthorized("invalid_user", "authenticated user is required") + } + if s.db == nil { + return teamManagerInfoRow{}, serviceUnavailable("database_unavailable", "database is unavailable") + } + + var row teamManagerInfoRow + err := s.db.WithContext(ctx). + Table("team_manager_base_info"). + Where("user_id = ?", user.UserID). + First(&row).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return teamManagerInfoRow{}, forbidden("not_manager", "user is not manager") + } + if err != nil { + return teamManagerInfoRow{}, serverError("manager_query_failed", err.Error()) + } + return row, nil +} + +func managerSysOrigin(manager teamManagerInfoRow, user AuthUser) string { + if origin := strings.TrimSpace(manager.Origin); origin != "" { + return origin + } + return strings.TrimSpace(user.SysOrigin) +} + +func (s *Service) userIsAdministrator(ctx context.Context, userID int64) (bool, error) { + var count int64 + if err := s.db.WithContext(ctx). + Table("sys_administrator"). + Where("user_id = ? AND status = ?", userID, true). + Count(&count).Error; err != nil { + return false, serverError("administrator_query_failed", err.Error()) + } + return count > 0, nil +} + +func (s *Service) userIsBDLeader(ctx context.Context, userID int64) (bool, error) { + var count int64 + if err := s.db.WithContext(ctx). + Table("room_bd_lead_base_info"). + Where("user_id = ?", userID). + Count(&count).Error; err != nil { + return false, serverError("bd_leader_query_failed", err.Error()) + } + return count > 0, nil +} + +func upsertBDBaseInfo(tx *gorm.DB, leader bdLeaderRow, operatorID int64, now time.Time) error { + var row businessDevelopmentBaseInfoRow + err := tx.Where("user_id = ?", leader.UserID).First(&row).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + id, idErr := utils.NextID() + if idErr != nil { + return serverError("id_generate_failed", idErr.Error()) + } + leadUserID := leader.UserID + return tx.Create(&businessDevelopmentBaseInfoRow{ + ID: id, + Origin: leader.Origin, + BDLeadUserID: &leadUserID, + UserID: leader.UserID, + Contact: leader.Contact, + Region: leader.RegionID, + MemberQuantity: 0, + CreateTime: now, + UpdateTime: now, + CreateUser: operatorID, + UpdateUser: operatorID, + }).Error + } + if err != nil { + return serverError("bd_info_query_failed", err.Error()) + } + if row.BDLeadUserID != nil && *row.BDLeadUserID > 0 { + return badRequest("user_already_bound_bd_leader", "user already has a BD leader") + } + + updates := map[string]any{ + "bd_lead_user_id": leader.UserID, + "region": leader.RegionID, + "update_time": now, + "update_user": operatorID, + } + if leader.Contact != "" { + updates["contact"] = leader.Contact + } + if leader.Origin != "" { + updates["sys_origin"] = leader.Origin + } + if err := tx.Model(&businessDevelopmentBaseInfoRow{}). + Where("id = ?", row.ID). + Updates(updates).Error; err != nil { + return serverError("bd_info_update_failed", err.Error()) + } + return nil +} + +func upsertUserHistoryBD(tx *gorm.DB, userID int64, operatorID int64, now time.Time) error { + var row userHistoryIdentityRow + err := tx.Where("user_id = ?", userID).First(&row).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return tx.Create(&userHistoryIdentityRow{ + UserID: userID, + IsBD: true, + CreateTime: now, + UpdateTime: now, + CreateUser: operatorID, + UpdateUser: operatorID, + }).Error + } + if err != nil { + return serverError("history_identity_query_failed", err.Error()) + } + if err := tx.Model(&userHistoryIdentityRow{}). + Where("user_id = ?", userID). + Updates(map[string]any{ + "is_bd": true, + "update_time": now, + "update_user": operatorID, + }).Error; err != nil { + return serverError("history_identity_update_failed", err.Error()) + } + return nil +} + +func bdLeaderRowToView(leader bdLeaderRow, user userBaseInfoRow) BDLeaderView { + return BDLeaderView{ + ID: ID(leader.ID), + UserID: ID(leader.UserID), + Account: user.Account, + UserAvatar: user.UserAvatar, + UserNickname: user.UserNickname, + Contact: leader.Contact, + RegionID: leader.RegionID, + OriginSys: leader.Origin, + CreateTime: leader.CreateTime, + UpdateTime: leader.UpdateTime, + } +} + +func bdLeaderListRowToView(row bdLeaderListRow) BDLeaderView { + return BDLeaderView{ + ID: ID(row.ID), + UserID: ID(row.UserID), + Account: row.Account, + UserAvatar: row.UserAvatar, + UserNickname: row.UserNickname, + Contact: row.Contact, + RegionID: row.RegionID, + OriginSys: row.Origin, + CreateTime: row.CreateTime, + UpdateTime: row.UpdateTime, + } +} diff --git a/internal/service/managercenter/service.go b/internal/service/managercenter/service.go index 20fcf3e..bf26597 100644 --- a/internal/service/managercenter/service.go +++ b/internal/service/managercenter/service.go @@ -14,9 +14,13 @@ import ( ) type teamManagerInfoRow struct { - ID int64 `gorm:"column:id;primaryKey"` - UserID int64 `gorm:"column:user_id"` - Origin string `gorm:"column:sys_origin"` + ID int64 `gorm:"column:id;primaryKey"` + UserID int64 `gorm:"column:user_id"` + Origin string `gorm:"column:sys_origin"` + Contact string `gorm:"column:contact"` + RegionID string `gorm:"column:region_id"` + CreateUser int64 `gorm:"column:create_user"` + UpdateUser int64 `gorm:"column:update_user"` } func (teamManagerInfoRow) TableName() string { @@ -324,25 +328,8 @@ func (s *Service) UnbanUser(ctx context.Context, user AuthUser, req UnbanUserReq } func (s *Service) ensureManager(ctx context.Context, user AuthUser) error { - if user.UserID <= 0 { - return unauthorized("invalid_user", "authenticated user is required") - } - if s.db == nil { - return serviceUnavailable("database_unavailable", "database is unavailable") - } - - var count int64 - err := s.db.WithContext(ctx). - Table("team_manager_base_info"). - Where("user_id = ?", user.UserID). - Count(&count).Error - if err != nil { - return serverError("manager_query_failed", err.Error()) - } - if count == 0 { - return forbidden("not_manager", "user is not manager") - } - return nil + _, err := s.requireManagerInfo(ctx, user) + return err } func (s *Service) findUserView(ctx context.Context, account string) (UserView, error) { diff --git a/internal/service/managercenter/service_test.go b/internal/service/managercenter/service_test.go index 527be9c..af32331 100644 --- a/internal/service/managercenter/service_test.go +++ b/internal/service/managercenter/service_test.go @@ -167,6 +167,73 @@ func TestGetProfileToleratesRechargeQueryFailure(t *testing.T) { } } +func TestAddBDLeaderCreatesLeaderAndListEntry(t *testing.T) { + service, _ := newManagerCenterTestService(t) + db := service.db.(*gorm.DB) + seedManager(t, db, 1) + seedUsers(t, db, + userBaseInfoRow{ID: 2, Account: "2002", UserNickname: "leader", OriginSys: "LIKEI"}, + ) + + resp, err := service.AddBDLeader(context.Background(), AuthUser{UserID: 1, SysOrigin: "LIKEI"}, AddBDLeaderRequest{ + UserID: ID(2), + Contact: "whatsapp", + }) + if err != nil { + t.Fatalf("AddBDLeader() error = %v", err) + } + if !resp.Success || resp.Leader.UserID.Int64() != 2 || resp.Leader.RegionID != "SA" { + t.Fatalf("resp = %+v, want created leader for user 2 in manager region", resp) + } + + var bdInfo businessDevelopmentBaseInfoRow + if err := db.Where("user_id = ?", int64(2)).First(&bdInfo).Error; err != nil { + t.Fatalf("load bd info: %v", err) + } + if bdInfo.BDLeadUserID == nil || *bdInfo.BDLeadUserID != 2 || bdInfo.Region != "SA" { + t.Fatalf("bd info = %+v, want self-linked BD leader info", bdInfo) + } + + var identity userHistoryIdentityRow + if err := db.Where("user_id = ?", int64(2)).First(&identity).Error; err != nil { + t.Fatalf("load identity: %v", err) + } + if !identity.IsBD { + t.Fatalf("identity = %+v, want is_bd=true", identity) + } + + list, err := service.ListBDLeaders(context.Background(), AuthUser{UserID: 1, SysOrigin: "LIKEI"}) + if err != nil { + t.Fatalf("ListBDLeaders() error = %v", err) + } + if list.Count != 1 || list.Records[0].UserID.Int64() != 2 || list.Records[0].Account != "2002" { + t.Fatalf("list = %+v, want one created leader", list) + } +} + +func TestAddBDLeaderRejectsExistingLeader(t *testing.T) { + service, _ := newManagerCenterTestService(t) + db := service.db.(*gorm.DB) + seedManager(t, db, 1) + seedUsers(t, db, userBaseInfoRow{ID: 2, Account: "2002", OriginSys: "LIKEI"}) + if err := db.Create(&bdLeaderRow{ID: 10, UserID: 2, Origin: "LIKEI", RegionID: "SA", CreateUser: 1}).Error; err != nil { + t.Fatalf("seed bd leader: %v", err) + } + + _, err := service.AddBDLeader(context.Background(), AuthUser{UserID: 1, SysOrigin: "LIKEI"}, AddBDLeaderRequest{UserID: ID(2)}) + assertAppErrorCode(t, err, "user_already_bd_leader") +} + +func TestAddBDLeaderRejectsTargetFromOtherOrigin(t *testing.T) { + service, _ := newManagerCenterTestService(t) + db := service.db.(*gorm.DB) + seedManager(t, db, 1) + seedUsers(t, db, userBaseInfoRow{ID: 2, Account: "2002", OriginSys: "ATYOU"}) + + _, err := service.AddBDLeader(context.Background(), AuthUser{UserID: 1, SysOrigin: "LIKEI"}, AddBDLeaderRequest{UserID: ID(2)}) + assertAppErrorCode(t, err, "bd_leader_origin_mismatch") +} + func TestSendPropsRejectsNonAdminFree(t *testing.T) { service, fake := newManagerCenterTestService(t) db := service.db.(*gorm.DB) @@ -480,6 +547,10 @@ func newManagerCenterTestService(t *testing.T) (*Service, *fakeManagerGateway) { if err := db.AutoMigrate( &teamManagerInfoRow{}, &userBaseInfoRow{}, + &bdLeaderRow{}, + &businessDevelopmentBaseInfoRow{}, + &userHistoryIdentityRow{}, + &administratorRow{}, &propsSourceRecordRow{}, &propsNobleVIPAbilityRow{}, &model.VipLevelConfig{}, @@ -498,7 +569,7 @@ func newManagerCenterTestService(t *testing.T) (*Service, *fakeManagerGateway) { func seedManager(t *testing.T, db *gorm.DB, userID int64) { t.Helper() - if err := db.Create(&teamManagerInfoRow{ID: userID, UserID: userID, Origin: "LIKEI"}).Error; err != nil { + if err := db.Create(&teamManagerInfoRow{ID: userID, UserID: userID, Origin: "LIKEI", RegionID: "SA"}).Error; err != nil { t.Fatalf("seed manager: %v", err) } } diff --git a/internal/service/managercenter/types.go b/internal/service/managercenter/types.go index 31af120..41eb216 100644 --- a/internal/service/managercenter/types.go +++ b/internal/service/managercenter/types.go @@ -6,6 +6,7 @@ import ( "net/http" "strconv" "strings" + "time" "chatapp3-golang/internal/common" "chatapp3-golang/internal/integration" @@ -178,6 +179,35 @@ type UnbanUserResponse struct { AccountStatus string `json:"accountStatus"` } +type BDLeaderListResponse struct { + Count int `json:"count"` + Records []BDLeaderView `json:"records"` +} + +type BDLeaderView struct { + ID ID `json:"id"` + UserID ID `json:"userId"` + Account string `json:"account"` + UserAvatar string `json:"userAvatar"` + UserNickname string `json:"userNickname"` + Contact string `json:"contact,omitempty"` + RegionID string `json:"regionId,omitempty"` + OriginSys string `json:"originSys,omitempty"` + CreateTime time.Time `json:"createTime,omitempty"` + UpdateTime time.Time `json:"updateTime,omitempty"` +} + +type AddBDLeaderRequest struct { + UserID ID `json:"userId"` + Contact string `json:"contact"` + RegionID string `json:"regionId"` +} + +type AddBDLeaderResponse struct { + Success bool `json:"success"` + Leader BDLeaderView `json:"leader"` +} + func badRequest(code, message string) error { return NewAppError(http.StatusBadRequest, code, message) } diff --git a/internal/service/regionimgroup/broadcast.go b/internal/service/regionimgroup/broadcast.go index 082370f..dbe6db7 100644 --- a/internal/service/regionimgroup/broadcast.go +++ b/internal/service/regionimgroup/broadcast.go @@ -67,6 +67,7 @@ func (s *Service) SendRegionBroadcast(ctx context.Context, req RegionBroadcastRe } } refreshHighWinMessage(data, messageType) + refreshCPRelationMessage(data, messageType) body := map[string]any{ "type": messageType, @@ -208,6 +209,25 @@ func highWinPlayTarget(messageType string) string { } } +func refreshCPRelationMessage(data map[string]any, messageType string) { + if strings.TrimSpace(messageType) != "CP_RELATION_BROADCAST" { + return + } + user := firstTextValue(data["userNickname"], data["nickname"], data["account"], data["actualAccount"], data["userId"]) + cpUser := firstTextValue(data["cpUserNickname"], data["acceptNickname"], data["toUserName"], data["cpUserAccount"], data["acceptAccount"], data["cpUserId"], data["acceptUserId"]) + if user == "" || cpUser == "" { + return + } + switch strings.ToUpper(strings.TrimSpace(firstTextValue(data["relationType"]))) { + case "BROTHER", "BROTHERS": + data["msg"] = fmt.Sprintf("%s与%s结为兄弟", user, cpUser) + case "SISTERS", "SISTER": + data["msg"] = fmt.Sprintf("%s与%s结为姐妹", user, cpUser) + default: + data["msg"] = fmt.Sprintf("%s与%s get cp relation", user, cpUser) + } +} + func firstTextValue(values ...any) string { for _, value := range values { text := strings.TrimSpace(fmt.Sprint(value)) diff --git a/internal/service/regionimgroup/service_test.go b/internal/service/regionimgroup/service_test.go index f2cdc5b..2250dad 100644 --- a/internal/service/regionimgroup/service_test.go +++ b/internal/service/regionimgroup/service_test.go @@ -179,6 +179,51 @@ func TestSendRegionBroadcastRefreshesHighWinMessageWithProfile(t *testing.T) { } } +func TestSendRegionBroadcastRefreshesCPRelationMessageWithProfiles(t *testing.T) { + service, db := newRegionIMGroupTestService(t) + fakeIM := &fakeIMGateway{} + service.im = fakeIM + now := time.Now() + if err := db.Create(&model.VoiceRoomRegionIMGroup{ + ID: 3011, + SysOrigin: "LIKEI", + RegionCode: "AR", + RegionName: "Arab", + GroupID: "@region-ar", + GroupName: "region-ar", + Status: imGroupStatusActive, + CreateTime: now, + UpdateTime: now, + }).Error; err != nil { + t.Fatalf("seed im group: %v", err) + } + + _, err := service.SendRegionBroadcast(context.Background(), RegionBroadcastRequest{ + SysOrigin: "LIKEI", + Type: "CP_RELATION_BROADCAST", + Data: map[string]any{ + "userId": "1001", + "acceptUserId": "1002", + "relationType": "SISTERS", + "msg": "1001与1002结为姐妹", + }, + }) + if err != nil { + t.Fatalf("SendRegionBroadcast() error = %v", err) + } + body, ok := fakeIM.sent[0].body.(map[string]any) + if !ok { + t.Fatalf("sent body type = %T", fakeIM.sent[0].body) + } + data, ok := body["data"].(map[string]any) + if !ok { + t.Fatalf("sent data type = %T", body["data"]) + } + if data["msg"] != "sender与claimer结为姐妹" { + t.Fatalf("msg = %v", data["msg"]) + } +} + func TestSendRegionBroadcastPrefersJavaRegionResolver(t *testing.T) { service, db := newRegionIMGroupTestService(t) service.regionResolver = fakeRegionResolver{