869 lines
37 KiB
Go
869 lines
37 KiB
Go
// Package cp 持久化 CP/兄弟/姐妹申请、单 active 关系、亲密值和等级默认规则。
|
||
package cp
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
cpdomain "hyapp/services/user-service/internal/domain/cp"
|
||
)
|
||
|
||
// Repository 是 CP 关系在 user-service MySQL 上的 owner 存储。
|
||
type Repository struct {
|
||
db *sql.DB
|
||
}
|
||
|
||
// New 基于共享 MySQL 连接池创建 CP 关系存储。
|
||
func New(db *sql.DB) *Repository {
|
||
return &Repository{db: db}
|
||
}
|
||
|
||
// ListApplications 按用户和方向读取申请，同时懒过期已超时的 pending 申请。
|
||
func (r *Repository) ListApplications(ctx context.Context, userID int64, direction string, status string, page int32, pageSize int32, nowMs int64) ([]cpdomain.Application, int64, error) {
|
||
if r == nil || r.db == nil {
|
||
return nil, 0, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
appCode := appcode.FromContext(ctx)
|
||
// 列表读取前先做一次轻量懒过期，避免客户端看到已经超过 24 小时但仍可操作的 pending 申请。
|
||
if _, err := r.db.ExecContext(ctx, `
|
||
UPDATE user_cp_applications
|
||
SET status = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND status = ? AND expires_at_ms <= ?`,
|
||
cpdomain.ApplicationStatusExpired, nowMs, appCode, cpdomain.ApplicationStatusPending, nowMs,
|
||
); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
|
||
where := []string{"a.app_code = ?"}
|
||
args := []any{appCode}
|
||
// direction 只影响 requester/target 过滤，不把双向申请合并，客户端才能区分我发出的和我收到的申请。
|
||
switch strings.ToLower(strings.TrimSpace(direction)) {
|
||
case "outgoing":
|
||
where = append(where, "a.requester_user_id = ?")
|
||
args = append(args, userID)
|
||
default:
|
||
where = append(where, "a.target_user_id = ?")
|
||
args = append(args, userID)
|
||
}
|
||
if status = strings.ToLower(strings.TrimSpace(status)); status != "" {
|
||
where = append(where, "a.status = ?")
|
||
args = append(args, status)
|
||
}
|
||
total, err := countRows(ctx, r.db, "SELECT COUNT(*) FROM user_cp_applications a WHERE "+strings.Join(where, " AND "), args...)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
args = append(args, clampPageSize(pageSize), offset(page, pageSize))
|
||
rows, err := r.db.QueryContext(ctx, applicationSelectSQL("WHERE "+strings.Join(where, " AND ")+`
|
||
ORDER BY a.created_at_ms DESC, a.application_id DESC
|
||
LIMIT ? OFFSET ?`), args...)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
defer rows.Close()
|
||
applications, err := scanApplications(rows)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
return applications, total, nil
|
||
}
|
||
|
||
// AcceptApplication 处理同意申请，事务内保证同一对用户只能形成一个 active 关系，并按关系类型容量限制个人数量。
|
||
func (r *Repository) AcceptApplication(ctx context.Context, userID int64, applicationID string, nowMs int64) (cpdomain.Application, cpdomain.Relationship, error) {
|
||
if r == nil || r.db == nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
appCode := appcode.FromContext(ctx)
|
||
if err := ensureDefaultConfigTx(ctx, tx, appCode, nowMs); err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
// 先锁申请行，再判断状态和过期时间，保证两个端同时点同意/拒绝时只有一个事务能改状态。
|
||
application, err := lockApplicationTx(ctx, tx, appCode, applicationID)
|
||
if err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
if application.Target.UserID != userID {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, xerr.New(xerr.PermissionDenied, "only target user can accept cp application")
|
||
}
|
||
if err := assertPendingApplication(application, nowMs); err != nil {
|
||
if isExpiredPendingApplication(application, nowMs) {
|
||
// 过期申请虽然不能继续同意，但 expired 本身是一个需要持久化的业务事实；
|
||
// 这里必须在返回冲突前提交事务，否则 defer Rollback 会把状态收敛一起回滚掉。
|
||
if markErr := markApplicationExpiredTx(ctx, tx, appCode, applicationID, nowMs); markErr != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, markErr
|
||
}
|
||
if commitErr := tx.Commit(); commitErr != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, commitErr
|
||
}
|
||
}
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
userAID, userBID := orderedPair(application.Requester.UserID, application.Target.UserID)
|
||
// A/B 关系按无序 pair 加锁；不论 CP、兄弟还是姐妹，同一时间只能存在一种 active 关系。
|
||
existing, found, err := lockActiveRelationshipTx(ctx, tx, appCode, userAID, userBID)
|
||
if err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
if found {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, xerr.New(xerr.Conflict, fmt.Sprintf("%s用户和你已经有%s关系", application.Requester.Username, existing.RelationType))
|
||
}
|
||
// 容量校验按关系类型读取后台配置；CP 固定只能有一个，兄弟/姐妹允许按运营配置扩展。
|
||
if err := assertRelationCapacityTx(ctx, tx, appCode, application.RelationType, application.Requester.UserID, application.Target.UserID); err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
|
||
relationshipID := relationshipIDForPair(userAID, userBID)
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO user_cp_relationships (
|
||
app_code, relationship_id, user_a_id, user_b_id, relation_type, status,
|
||
intimacy_value, level_no, source_application_id,
|
||
formed_at_ms, ended_at_ms, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, 0, 1, ?, ?, 0, ?, ?)`,
|
||
appCode, relationshipID, userAID, userBID, application.RelationType, cpdomain.RelationshipStatusActive,
|
||
application.ApplicationID, nowMs, nowMs, nowMs,
|
||
); err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
// 同意一种关系后，同一对用户之间其他类型的 pending 申请不再允许接受，状态置为 blocked 便于客户端提示。
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE user_cp_applications
|
||
SET status = ?, decided_at_ms = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND application_id = ?`,
|
||
cpdomain.ApplicationStatusAccepted, nowMs, nowMs, appCode, application.ApplicationID,
|
||
); err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE user_cp_applications
|
||
SET status = ?, updated_at_ms = ?
|
||
WHERE app_code = ?
|
||
AND status = ?
|
||
AND requester_user_id IN (?, ?)
|
||
AND target_user_id IN (?, ?)
|
||
AND application_id <> ?`,
|
||
cpdomain.ApplicationStatusBlocked, nowMs, appCode, cpdomain.ApplicationStatusPending,
|
||
application.Requester.UserID, application.Target.UserID, application.Requester.UserID, application.Target.UserID, application.ApplicationID,
|
||
); err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
application.Status = cpdomain.ApplicationStatusAccepted
|
||
application.UpdatedAtMS = nowMs
|
||
application.DecidedAtMS = nowMs
|
||
relationship, err := queryRelationshipByIDTx(ctx, tx, appCode, userID, relationshipID)
|
||
if err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
// user_outbox 只记录事实；IM、区域广播和后续 push 由下游 notice/统计等消费者按事件类型各自投递。
|
||
if err := insertOutboxTx(ctx, tx, appCode, cpdomain.EventTypeApplicationAccepted, "cp_application", application.Target.UserID, application, nowMs); err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
relationshipPayload := map[string]any{"application": application, "relationship": relationship}
|
||
if err := insertOutboxTx(ctx, tx, appCode, cpdomain.EventTypeRelationshipCreated, "cp_relationship", application.Target.UserID, relationshipPayload, nowMs); err != nil {
|
||
return cpdomain.Application{}, cpdomain.Relationship{}, err
|
||
}
|
||
return application, relationship, tx.Commit()
|
||
}
|
||
|
||
// RejectApplication 处理拒绝申请；拒绝只改变该申请，不影响同一对用户其他 pending 类型。
|
||
func (r *Repository) RejectApplication(ctx context.Context, userID int64, applicationID string, reason string, nowMs int64) (cpdomain.Application, error) {
|
||
if r == nil || r.db == nil {
|
||
return cpdomain.Application{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return cpdomain.Application{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
appCode := appcode.FromContext(ctx)
|
||
application, err := lockApplicationTx(ctx, tx, appCode, applicationID)
|
||
if err != nil {
|
||
return cpdomain.Application{}, err
|
||
}
|
||
if application.Target.UserID != userID {
|
||
return cpdomain.Application{}, xerr.New(xerr.PermissionDenied, "only target user can reject cp application")
|
||
}
|
||
if err := assertPendingApplication(application, nowMs); err != nil {
|
||
if isExpiredPendingApplication(application, nowMs) {
|
||
// 拒绝接口同样承担惰性过期收敛；过期事实要先提交，再把“不能操作”的冲突返回给客户端。
|
||
if markErr := markApplicationExpiredTx(ctx, tx, appCode, applicationID, nowMs); markErr != nil {
|
||
return cpdomain.Application{}, markErr
|
||
}
|
||
if commitErr := tx.Commit(); commitErr != nil {
|
||
return cpdomain.Application{}, commitErr
|
||
}
|
||
}
|
||
return cpdomain.Application{}, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE user_cp_applications
|
||
SET status = ?, decided_at_ms = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND application_id = ?`,
|
||
cpdomain.ApplicationStatusRejected, nowMs, nowMs, appCode, application.ApplicationID,
|
||
); err != nil {
|
||
return cpdomain.Application{}, err
|
||
}
|
||
application.Status = cpdomain.ApplicationStatusRejected
|
||
application.UpdatedAtMS = nowMs
|
||
application.DecidedAtMS = nowMs
|
||
payload := map[string]any{"application": application, "reason": strings.TrimSpace(reason)}
|
||
if err := insertOutboxTx(ctx, tx, appCode, cpdomain.EventTypeApplicationRejected, "cp_application", application.Target.UserID, payload, nowMs); err != nil {
|
||
return cpdomain.Application{}, err
|
||
}
|
||
return application, tx.Commit()
|
||
}
|
||
|
||
// ListRelationships 读取用户 active 关系列表；CP 最多一条，兄弟/姐妹可能多条，因此分页必须保留。
|
||
func (r *Repository) ListRelationships(ctx context.Context, userID int64, relationType string, page int32, pageSize int32) ([]cpdomain.Relationship, int64, error) {
|
||
if r == nil || r.db == nil {
|
||
return nil, 0, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
appCode := appcode.FromContext(ctx)
|
||
where := []string{"r.app_code = ?", "r.status = ?", "(r.user_a_id = ? OR r.user_b_id = ?)"}
|
||
args := []any{appCode, cpdomain.RelationshipStatusActive, userID, userID}
|
||
if relationType = normalizeRelationType(relationType); relationType != "" {
|
||
where = append(where, "r.relation_type = ?")
|
||
args = append(args, relationType)
|
||
}
|
||
// total 只统计 active 关系；历史关系不会占用当前关系数量，也不会在 App 资料卡展示。
|
||
total, err := countRows(ctx, r.db, "SELECT COUNT(*) FROM user_cp_relationships r WHERE "+strings.Join(where, " AND "), args...)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
// relationshipSelectSQL 的前两个占位是 viewer/me 和 partner 选择，suffix 的过滤参数必须跟在它们后面。
|
||
queryArgs := []any{userID, userID}
|
||
queryArgs = append(queryArgs, args...)
|
||
queryArgs = append(queryArgs, clampPageSize(pageSize), offset(page, pageSize))
|
||
rows, err := r.db.QueryContext(ctx, relationshipSelectSQL("WHERE "+strings.Join(where, " AND ")+`
|
||
ORDER BY r.formed_at_ms DESC, r.relationship_id DESC
|
||
LIMIT ? OFFSET ?`), queryArgs...)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
defer rows.Close()
|
||
relationships, err := scanRelationships(rows)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
return relationships, total, nil
|
||
}
|
||
|
||
// ConsumeGiftEvent 把房间送礼事实转换为 CP 申请或亲密值增长。
|
||
func (r *Repository) ConsumeGiftEvent(ctx context.Context, event cpdomain.GiftEvent, nowMs int64) (cpdomain.ConsumeResult, error) {
|
||
if r == nil || r.db == nil {
|
||
return cpdomain.ConsumeResult{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
relationType := normalizeRelationType(event.CPRelationType)
|
||
if relationType == "" || event.SenderUserID <= 0 || event.TargetUserID <= 0 || event.SenderUserID == event.TargetUserID {
|
||
return cpdomain.ConsumeResult{Consumed: false}, nil
|
||
}
|
||
if strings.TrimSpace(event.EventID) == "" {
|
||
return cpdomain.ConsumeResult{}, xerr.New(xerr.InvalidArgument, "room event id is required")
|
||
}
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return cpdomain.ConsumeResult{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
appCode := appcode.FromContext(ctx)
|
||
if event.AppCode != "" {
|
||
appCode = appcode.Normalize(event.AppCode)
|
||
}
|
||
if err := ensureDefaultConfigTx(ctx, tx, appCode, nowMs); err != nil {
|
||
return cpdomain.ConsumeResult{}, err
|
||
}
|
||
// room outbox event_id 是消费幂等键；重复投递时直接提交空事务，避免重复创建申请或重复加亲密值。
|
||
if exists, err := lockConsumedEventTx(ctx, tx, appCode, event.EventID); err != nil {
|
||
return cpdomain.ConsumeResult{}, err
|
||
} else if exists {
|
||
return cpdomain.ConsumeResult{Consumed: false}, tx.Commit()
|
||
}
|
||
userAID, userBID := orderedPair(event.SenderUserID, event.TargetUserID)
|
||
// 已有 active 关系时，任何关系礼物都只增加这对关系的亲密值，不再生成新的申请。
|
||
if relationship, found, err := lockActiveRelationshipTx(ctx, tx, appCode, userAID, userBID); err != nil {
|
||
return cpdomain.ConsumeResult{}, err
|
||
} else if found {
|
||
updated, err := addRelationshipIntimacyTx(ctx, tx, appCode, relationship, event.GiftValue, event.TargetUserID, nowMs)
|
||
if err != nil {
|
||
return cpdomain.ConsumeResult{}, err
|
||
}
|
||
if err := insertConsumptionTx(ctx, tx, appCode, event.EventID, "relationship_intimacy_added", "", updated.RelationshipID, "", nowMs); err != nil {
|
||
return cpdomain.ConsumeResult{}, err
|
||
}
|
||
if err := insertOutboxTx(ctx, tx, appCode, cpdomain.EventTypeRelationshipIntimacyChanged, "cp_relationship", event.TargetUserID, map[string]any{"relationship": updated, "gift_event": event}, nowMs); err != nil {
|
||
return cpdomain.ConsumeResult{}, err
|
||
}
|
||
return cpdomain.ConsumeResult{Consumed: true, Relationship: updated}, tx.Commit()
|
||
}
|
||
// 没有 active 关系时，CP/兄弟/姐妹礼物只生成 pending 申请；真正建立关系必须由 B 显式同意。
|
||
if err := ensurePairHasNoActiveRelationshipTx(ctx, tx, appCode, userAID, userBID, event.TargetUserID); err != nil {
|
||
if err := insertConsumptionTx(ctx, tx, appCode, event.EventID, "skipped", "", "", "existing_relationship", 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
|
||
}
|
||
if err := insertConsumptionTx(ctx, tx, appCode, event.EventID, "application_created", application.ApplicationID, "", "", nowMs); err != nil {
|
||
return cpdomain.ConsumeResult{}, err
|
||
}
|
||
if err := insertOutboxTx(ctx, tx, appCode, cpdomain.EventTypeApplicationCreated, "cp_application", application.Target.UserID, application, nowMs); err != nil {
|
||
return cpdomain.ConsumeResult{}, err
|
||
}
|
||
return cpdomain.ConsumeResult{Consumed: true, Application: application}, tx.Commit()
|
||
}
|
||
|
||
func ensureDefaultConfigTx(ctx context.Context, tx *sql.Tx, appCode string, nowMs int64) error {
|
||
// 默认值只兜底本地和新 App 启动；CP 固定 1 个，兄弟/姐妹默认 5 个并允许后台覆盖。
|
||
for _, relationType := range []string{cpdomain.RelationTypeCP, cpdomain.RelationTypeBrother, cpdomain.RelationTypeSister} {
|
||
maxCount := int32(1)
|
||
if relationType == cpdomain.RelationTypeBrother || relationType == cpdomain.RelationTypeSister {
|
||
maxCount = 5
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO user_cp_relation_configs (
|
||
app_code, relation_type, max_count_per_user, application_expire_hours,
|
||
status, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, 24, 'active', ?, ?)
|
||
ON DUPLICATE KEY UPDATE relation_type = VALUES(relation_type)`,
|
||
appCode, relationType, maxCount, nowMs, nowMs,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
for level := int32(1); level <= 5; level++ {
|
||
threshold := int64(level-1) * 10000
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO user_cp_level_rules (
|
||
app_code, relation_type, level_no, intimacy_threshold,
|
||
reward_resource_group_id, level_icon_url, status, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, 0, '', 'active', ?, ?)
|
||
ON DUPLICATE KEY UPDATE level_no = VALUES(level_no)`,
|
||
appCode, relationType, level, threshold, nowMs, nowMs,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func applicationSelectSQL(suffix string) string {
|
||
// 申请查询统一 join 双方 users 快照，避免 service 层为每条申请再发起批量用户查询。
|
||
return `
|
||
SELECT
|
||
a.application_id, a.relation_type, a.status,
|
||
a.room_id, a.room_region_id,
|
||
a.gift_id, a.gift_name, a.gift_icon_url, a.gift_animation_url,
|
||
a.gift_count, a.gift_value, a.billing_receipt_id,
|
||
a.source_room_event_id, a.source_command_id,
|
||
a.created_at_ms, a.updated_at_ms, a.expires_at_ms, a.decided_at_ms,
|
||
requester.user_id, COALESCE(requester.current_display_user_id, ''), COALESCE(requester.username, ''), COALESCE(requester.avatar, ''),
|
||
target.user_id, COALESCE(target.current_display_user_id, ''), COALESCE(target.username, ''), COALESCE(target.avatar, '')
|
||
FROM user_cp_applications a
|
||
JOIN users requester ON requester.app_code = a.app_code AND requester.user_id = a.requester_user_id
|
||
JOIN users target ON target.app_code = a.app_code AND target.user_id = a.target_user_id
|
||
` + suffix
|
||
}
|
||
|
||
func scanApplications(rows *sql.Rows) ([]cpdomain.Application, error) {
|
||
applications := make([]cpdomain.Application, 0)
|
||
for rows.Next() {
|
||
application, err := scanApplication(rows)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
applications = append(applications, application)
|
||
}
|
||
return applications, rows.Err()
|
||
}
|
||
|
||
func scanApplication(scanner interface{ Scan(dest ...any) error }) (cpdomain.Application, error) {
|
||
var app cpdomain.Application
|
||
err := scanner.Scan(
|
||
&app.ApplicationID, &app.RelationType, &app.Status,
|
||
&app.RoomID, &app.RoomRegionID,
|
||
&app.Gift.GiftID, &app.Gift.GiftName, &app.Gift.GiftIconURL, &app.Gift.GiftAnimationURL,
|
||
&app.Gift.GiftCount, &app.Gift.GiftValue, &app.Gift.BillingReceiptID,
|
||
&app.SourceEventID, &app.SourceCommandID,
|
||
&app.CreatedAtMS, &app.UpdatedAtMS, &app.ExpiresAtMS, &app.DecidedAtMS,
|
||
&app.Requester.UserID, &app.Requester.DisplayUserID, &app.Requester.Username, &app.Requester.Avatar,
|
||
&app.Target.UserID, &app.Target.DisplayUserID, &app.Target.Username, &app.Target.Avatar,
|
||
)
|
||
return app, err
|
||
}
|
||
|
||
func lockApplicationTx(ctx context.Context, tx *sql.Tx, appCode string, applicationID string) (cpdomain.Application, error) {
|
||
row := tx.QueryRowContext(ctx, applicationSelectSQL(`
|
||
WHERE a.app_code = ? AND a.application_id = ?
|
||
FOR UPDATE`), appCode, strings.TrimSpace(applicationID))
|
||
application, err := scanApplication(row)
|
||
if err == sql.ErrNoRows {
|
||
return cpdomain.Application{}, xerr.New(xerr.NotFound, "cp application not found")
|
||
}
|
||
return application, err
|
||
}
|
||
|
||
func markApplicationExpiredTx(ctx context.Context, tx *sql.Tx, appCode string, applicationID string, nowMs int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
UPDATE user_cp_applications
|
||
SET status = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND application_id = ? AND status = ?`,
|
||
cpdomain.ApplicationStatusExpired, nowMs, appCode, applicationID, cpdomain.ApplicationStatusPending,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func isExpiredPendingApplication(application cpdomain.Application, nowMs int64) bool {
|
||
// 只有 pending 且超过过期时间的申请才做惰性状态收敛；accepted/rejected/blocked 等终态不能被误改成 expired。
|
||
return application.Status == cpdomain.ApplicationStatusPending && application.ExpiresAtMS <= nowMs
|
||
}
|
||
|
||
func assertPendingApplication(application cpdomain.Application, nowMs int64) error {
|
||
if application.Status != cpdomain.ApplicationStatusPending {
|
||
return xerr.New(xerr.Conflict, "cp application is not pending")
|
||
}
|
||
if application.ExpiresAtMS <= nowMs {
|
||
return xerr.New(xerr.Conflict, "cp application expired")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func relationshipSelectSQL(suffix string) string {
|
||
// viewerUserID 连续出现两次：me 投影和 partner 选择都必须以当前查看用户为准。
|
||
// 等级阈值也在查询时投影出来，客户端资料卡只能使用服务端计算后的进度，不能读取后台配置或硬编码等级表。
|
||
return `
|
||
SELECT
|
||
r.relationship_id, r.relation_type, r.status, r.user_a_id, r.user_b_id,
|
||
r.intimacy_value, r.level_no, r.source_application_id, r.formed_at_ms, r.updated_at_ms,
|
||
COALESCE((
|
||
SELECT current_rule.intimacy_threshold
|
||
FROM user_cp_level_rules current_rule
|
||
WHERE current_rule.app_code = r.app_code
|
||
AND current_rule.relation_type = r.relation_type
|
||
AND current_rule.level_no = r.level_no
|
||
AND current_rule.status = 'active'
|
||
LIMIT 1
|
||
), 0) AS current_level_threshold,
|
||
COALESCE((
|
||
SELECT next_rule.intimacy_threshold
|
||
FROM user_cp_level_rules next_rule
|
||
WHERE next_rule.app_code = r.app_code
|
||
AND next_rule.relation_type = r.relation_type
|
||
AND next_rule.level_no > r.level_no
|
||
AND next_rule.status = 'active'
|
||
ORDER BY next_rule.level_no ASC
|
||
LIMIT 1
|
||
), 0) AS next_level_threshold,
|
||
me.user_id, COALESCE(me.current_display_user_id, ''), COALESCE(me.username, ''), COALESCE(me.avatar, ''),
|
||
partner.user_id, COALESCE(partner.current_display_user_id, ''), COALESCE(partner.username, ''), COALESCE(partner.avatar, '')
|
||
FROM user_cp_relationships r
|
||
JOIN users me ON me.app_code = r.app_code AND me.user_id = ?
|
||
JOIN users partner ON partner.app_code = r.app_code
|
||
AND partner.user_id = CASE WHEN r.user_a_id = ? THEN r.user_b_id ELSE r.user_a_id END
|
||
` + suffix
|
||
}
|
||
|
||
func scanRelationships(rows *sql.Rows) ([]cpdomain.Relationship, error) {
|
||
relationships := make([]cpdomain.Relationship, 0)
|
||
for rows.Next() {
|
||
relationship, err := scanRelationship(rows)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
relationships = append(relationships, relationship)
|
||
}
|
||
return relationships, rows.Err()
|
||
}
|
||
|
||
func scanRelationship(scanner interface{ Scan(dest ...any) error }) (cpdomain.Relationship, error) {
|
||
var relationship cpdomain.Relationship
|
||
err := scanner.Scan(
|
||
&relationship.RelationshipID, &relationship.RelationType, &relationship.Status, &relationship.UserAID, &relationship.UserBID,
|
||
&relationship.IntimacyValue, &relationship.Level, &relationship.SourceApplicationID, &relationship.FormedAtMS, &relationship.UpdatedAtMS,
|
||
&relationship.CurrentLevelThreshold, &relationship.NextLevelThreshold,
|
||
&relationship.Me.UserID, &relationship.Me.DisplayUserID, &relationship.Me.Username, &relationship.Me.Avatar,
|
||
&relationship.Partner.UserID, &relationship.Partner.DisplayUserID, &relationship.Partner.Username, &relationship.Partner.Avatar,
|
||
)
|
||
if err == nil {
|
||
applyRelationshipLevelProgress(&relationship)
|
||
}
|
||
return relationship, err
|
||
}
|
||
|
||
func applyRelationshipLevelProgress(relationship *cpdomain.Relationship) {
|
||
if relationship == nil {
|
||
return
|
||
}
|
||
if relationship.CurrentLevelThreshold < 0 {
|
||
relationship.CurrentLevelThreshold = 0
|
||
}
|
||
if relationship.NextLevelThreshold <= relationship.CurrentLevelThreshold {
|
||
// 没有下一等级规则时按满级处理；客户端可以直接隐藏“距离下一级”和进度条。
|
||
relationship.NextLevelThreshold = 0
|
||
relationship.NeededForNextLevel = 0
|
||
relationship.LevelProgressPercent = 100
|
||
relationship.MaxLevel = true
|
||
return
|
||
}
|
||
remaining := relationship.NextLevelThreshold - relationship.IntimacyValue
|
||
if remaining < 0 {
|
||
remaining = 0
|
||
}
|
||
relationship.NeededForNextLevel = remaining
|
||
span := relationship.NextLevelThreshold - relationship.CurrentLevelThreshold
|
||
filled := relationship.IntimacyValue - relationship.CurrentLevelThreshold
|
||
if filled < 0 {
|
||
filled = 0
|
||
}
|
||
progress := int32((filled * 100) / span)
|
||
if progress < 0 {
|
||
progress = 0
|
||
}
|
||
if progress > 100 {
|
||
progress = 100
|
||
}
|
||
relationship.LevelProgressPercent = progress
|
||
relationship.MaxLevel = false
|
||
}
|
||
|
||
func queryRelationshipByIDTx(ctx context.Context, tx *sql.Tx, appCode string, viewerUserID int64, relationshipID string) (cpdomain.Relationship, error) {
|
||
row := tx.QueryRowContext(ctx, relationshipSelectSQL(`
|
||
WHERE r.app_code = ? AND r.relationship_id = ?`),
|
||
viewerUserID, viewerUserID, appCode, strings.TrimSpace(relationshipID),
|
||
)
|
||
relationship, err := scanRelationship(row)
|
||
if err == sql.ErrNoRows {
|
||
return cpdomain.Relationship{}, xerr.New(xerr.NotFound, "cp relationship not found")
|
||
}
|
||
return relationship, err
|
||
}
|
||
|
||
func lockActiveRelationshipTx(ctx context.Context, tx *sql.Tx, appCode string, userAID int64, userBID int64) (cpdomain.Relationship, bool, error) {
|
||
// active 关系行用 FOR UPDATE 锁住 pair，关系创建和送礼加亲密值都走这把锁。
|
||
row := tx.QueryRowContext(ctx, `
|
||
SELECT relationship_id, relation_type, status, user_a_id, user_b_id,
|
||
intimacy_value, level_no, source_application_id, formed_at_ms, updated_at_ms
|
||
FROM user_cp_relationships
|
||
WHERE app_code = ? AND user_a_id = ? AND user_b_id = ? AND status = ?
|
||
FOR UPDATE`,
|
||
appCode, userAID, userBID, cpdomain.RelationshipStatusActive,
|
||
)
|
||
var relationship cpdomain.Relationship
|
||
err := row.Scan(
|
||
&relationship.RelationshipID, &relationship.RelationType, &relationship.Status, &relationship.UserAID, &relationship.UserBID,
|
||
&relationship.IntimacyValue, &relationship.Level, &relationship.SourceApplicationID, &relationship.FormedAtMS, &relationship.UpdatedAtMS,
|
||
)
|
||
if err == sql.ErrNoRows {
|
||
return cpdomain.Relationship{}, false, nil
|
||
}
|
||
return relationship, true, err
|
||
}
|
||
|
||
func ensurePairHasNoActiveRelationshipTx(ctx context.Context, tx *sql.Tx, appCode string, userAID int64, userBID int64, viewerUserID int64) error {
|
||
if relationship, found, err := lockActiveRelationshipTx(ctx, tx, appCode, userAID, userBID); err != nil {
|
||
return err
|
||
} else if found {
|
||
partnerID := userAID
|
||
if partnerID == viewerUserID {
|
||
partnerID = userBID
|
||
}
|
||
return xerr.New(xerr.Conflict, fmt.Sprintf("%d用户和你已经有%s关系", partnerID, relationship.RelationType))
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func assertRelationCapacityTx(ctx context.Context, tx *sql.Tx, appCode string, relationType string, userAID int64, userBID int64) error {
|
||
if normalizeRelationType(relationType) == "" {
|
||
return xerr.New(xerr.Conflict, "cp relation type is disabled")
|
||
}
|
||
maxCount, err := relationMaxCountTx(ctx, tx, appCode, relationType)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, userID := range []int64{userAID, userBID} {
|
||
// 容量按用户和关系类型计算；同一对用户互斥由 pair 锁保证，不同对象可以同时拥有兄弟/姐妹关系。
|
||
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,
|
||
)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if total >= int64(maxCount) {
|
||
return xerr.New(xerr.Conflict, "user cp relation count reaches limit")
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func relationMaxCountTx(ctx context.Context, tx *sql.Tx, appCode string, relationType string) (int32, error) {
|
||
// CP 的“一人只能有一个”是产品硬约束，不能被后台或脏数据配置放宽。
|
||
if normalizeRelationType(relationType) == cpdomain.RelationTypeCP {
|
||
return 1, nil
|
||
}
|
||
var maxCount int32
|
||
err := tx.QueryRowContext(ctx, `
|
||
SELECT max_count_per_user
|
||
FROM user_cp_relation_configs
|
||
WHERE app_code = ? AND relation_type = ? AND status = 'active'
|
||
FOR UPDATE`, appCode, relationType).Scan(&maxCount)
|
||
if err == sql.ErrNoRows {
|
||
return 0, xerr.New(xerr.Conflict, "cp relation type is disabled")
|
||
}
|
||
if maxCount <= 0 {
|
||
maxCount = 1
|
||
}
|
||
return maxCount, err
|
||
}
|
||
|
||
func addRelationshipIntimacyTx(ctx context.Context, tx *sql.Tx, appCode string, relationship cpdomain.Relationship, delta int64, viewerUserID int64, nowMs int64) (cpdomain.Relationship, error) {
|
||
if delta < 0 {
|
||
delta = 0
|
||
}
|
||
nextValue := relationship.IntimacyValue + delta
|
||
// 等级直接由当前亲密值重算，避免漏处理一次大额礼物跨多级升级。
|
||
nextLevel, err := levelForIntimacyTx(ctx, tx, appCode, relationship.RelationType, nextValue)
|
||
if err != nil {
|
||
return cpdomain.Relationship{}, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE user_cp_relationships
|
||
SET intimacy_value = ?, level_no = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND relationship_id = ?`,
|
||
nextValue, nextLevel, nowMs, appCode, relationship.RelationshipID,
|
||
); err != nil {
|
||
return cpdomain.Relationship{}, err
|
||
}
|
||
return queryRelationshipByIDTx(ctx, tx, appCode, viewerUserID, relationship.RelationshipID)
|
||
}
|
||
|
||
func levelForIntimacyTx(ctx context.Context, tx *sql.Tx, appCode string, relationType string, intimacy int64) (int32, error) {
|
||
rows, err := tx.QueryContext(ctx, `
|
||
SELECT level_no, intimacy_threshold
|
||
FROM user_cp_level_rules
|
||
WHERE app_code = ? AND relation_type = ? AND status = 'active'
|
||
ORDER BY level_no ASC`, appCode, relationType)
|
||
if err != nil {
|
||
return 1, err
|
||
}
|
||
defer rows.Close()
|
||
level := int32(1)
|
||
for rows.Next() {
|
||
var candidate int32
|
||
var threshold int64
|
||
if err := rows.Scan(&candidate, &threshold); err != nil {
|
||
return 1, err
|
||
}
|
||
if intimacy >= threshold {
|
||
level = candidate
|
||
}
|
||
}
|
||
return level, rows.Err()
|
||
}
|
||
|
||
func upsertApplicationFromGiftTx(ctx context.Context, tx *sql.Tx, appCode string, event cpdomain.GiftEvent, relationType string, nowMs int64) (cpdomain.Application, error) {
|
||
expireHours, err := applicationExpireHoursTx(ctx, tx, appCode, relationType)
|
||
if err != nil {
|
||
return cpdomain.Application{}, err
|
||
}
|
||
expiresAtMS := nowMs + int64(expireHours)*time.Hour.Milliseconds()
|
||
row := tx.QueryRowContext(ctx, applicationSelectSQL(`
|
||
WHERE a.app_code = ?
|
||
AND a.requester_user_id = ?
|
||
AND a.target_user_id = ?
|
||
AND a.relation_type = ?
|
||
AND a.status = ?
|
||
AND a.expires_at_ms > ?
|
||
ORDER BY a.created_at_ms DESC
|
||
LIMIT 1
|
||
FOR UPDATE`),
|
||
appCode, event.SenderUserID, event.TargetUserID, relationType, cpdomain.ApplicationStatusPending, nowMs,
|
||
)
|
||
application, err := scanApplication(row)
|
||
if err != nil && err != sql.ErrNoRows {
|
||
return cpdomain.Application{}, err
|
||
}
|
||
if err == nil {
|
||
// 同一对用户、同一类型已有 pending 申请时刷新礼物快照和过期时间，不生成多条待处理卡片。
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE user_cp_applications
|
||
SET room_id = ?, room_region_id = ?, gift_id = ?, gift_name = ?, gift_icon_url = ?,
|
||
gift_animation_url = ?, gift_count = ?, gift_value = ?, billing_receipt_id = ?,
|
||
source_room_event_id = ?, source_command_id = ?, expires_at_ms = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND application_id = ?`,
|
||
event.RoomID, event.VisibleRegionID, event.GiftID, event.GiftName, event.GiftIconURL,
|
||
event.GiftAnimationURL, event.GiftCount, event.GiftValue, event.BillingReceiptID,
|
||
event.EventID, event.CommandID, expiresAtMS, nowMs, appCode, application.ApplicationID,
|
||
); err != nil {
|
||
return cpdomain.Application{}, err
|
||
}
|
||
application.RoomID = event.RoomID
|
||
application.RoomRegionID = event.VisibleRegionID
|
||
application.Gift = giftSnapshotFromEvent(event)
|
||
application.SourceEventID = event.EventID
|
||
application.SourceCommandID = event.CommandID
|
||
application.ExpiresAtMS = expiresAtMS
|
||
application.UpdatedAtMS = nowMs
|
||
return application, nil
|
||
}
|
||
applicationID := fmt.Sprintf("cp_app:%s", strings.TrimSpace(event.EventID))
|
||
// 新申请 ID 派生自 room outbox event_id，结合消费位点可以保证 MQ 重投不会生成第二条申请。
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO user_cp_applications (
|
||
app_code, application_id, requester_user_id, target_user_id, relation_type, status,
|
||
room_id, room_region_id, gift_id, gift_name, gift_icon_url, gift_animation_url,
|
||
gift_count, gift_value, billing_receipt_id, source_room_event_id, source_command_id,
|
||
created_at_ms, updated_at_ms, expires_at_ms, decided_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0)`,
|
||
appCode, applicationID, event.SenderUserID, event.TargetUserID, relationType, cpdomain.ApplicationStatusPending,
|
||
event.RoomID, event.VisibleRegionID, event.GiftID, event.GiftName, event.GiftIconURL, event.GiftAnimationURL,
|
||
event.GiftCount, event.GiftValue, event.BillingReceiptID, event.EventID, event.CommandID,
|
||
nowMs, nowMs, expiresAtMS,
|
||
); err != nil {
|
||
return cpdomain.Application{}, err
|
||
}
|
||
row = tx.QueryRowContext(ctx, applicationSelectSQL(`
|
||
WHERE a.app_code = ? AND a.application_id = ?`), appCode, applicationID)
|
||
return scanApplication(row)
|
||
}
|
||
|
||
func applicationExpireHoursTx(ctx context.Context, tx *sql.Tx, appCode string, relationType string) (int32, error) {
|
||
var hours int32
|
||
err := tx.QueryRowContext(ctx, `
|
||
SELECT application_expire_hours
|
||
FROM user_cp_relation_configs
|
||
WHERE app_code = ? AND relation_type = ? AND status = 'active'`, appCode, relationType).Scan(&hours)
|
||
if err == sql.ErrNoRows {
|
||
return 0, xerr.New(xerr.Conflict, "cp relation type is disabled")
|
||
}
|
||
if hours <= 0 {
|
||
hours = 24
|
||
}
|
||
return hours, err
|
||
}
|
||
|
||
func lockConsumedEventTx(ctx context.Context, tx *sql.Tx, appCode string, eventID string) (bool, error) {
|
||
// 消费位点也加锁，防止同一 event_id 在并行 consumer 中同时进入创建/加亲密值分支。
|
||
var found string
|
||
err := tx.QueryRowContext(ctx, `
|
||
SELECT room_event_id
|
||
FROM user_cp_gift_event_consumption
|
||
WHERE app_code = ? AND room_event_id = ?
|
||
FOR UPDATE`, appCode, eventID).Scan(&found)
|
||
if err == sql.ErrNoRows {
|
||
return false, nil
|
||
}
|
||
return err == nil, err
|
||
}
|
||
|
||
func insertConsumptionTx(ctx context.Context, tx *sql.Tx, appCode string, eventID string, status string, applicationID string, relationshipID string, skipReason string, nowMs int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO user_cp_gift_event_consumption (
|
||
app_code, room_event_id, status, application_id, relationship_id,
|
||
skip_reason, consumed_at_ms, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
appCode, eventID, status, applicationID, relationshipID, skipReason, nowMs, nowMs, nowMs,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func insertOutboxTx(ctx context.Context, tx *sql.Tx, appCode string, eventType string, aggregateType string, aggregateID int64, payload any, nowMs int64) error {
|
||
payloadBytes, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
eventID := fmt.Sprintf("%s:%s:%d:%d", aggregateType, eventType, aggregateID, nowMs)
|
||
// outbox 与业务状态同事务提交；RocketMQ 发布失败只影响异步 IM/统计，不回滚已经同意的关系。
|
||
_, err = tx.ExecContext(ctx, `
|
||
INSERT INTO user_outbox (
|
||
app_code, event_id, event_type, aggregate_type, aggregate_id,
|
||
status, worker_id, lock_until_ms, retry_count, next_retry_at_ms,
|
||
last_error, payload_json, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, 'pending', '', 0, 0, 0, '', CAST(? AS JSON), ?, ?)
|
||
ON DUPLICATE KEY UPDATE event_id = VALUES(event_id)`,
|
||
appCode, eventID, eventType, aggregateType, aggregateID, string(payloadBytes), nowMs, nowMs,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func countRows(ctx context.Context, q interface {
|
||
QueryRowContext(context.Context, string, ...any) *sql.Row
|
||
}, query string, args ...any) (int64, error) {
|
||
var total int64
|
||
err := q.QueryRowContext(ctx, query, args...).Scan(&total)
|
||
return total, err
|
||
}
|
||
|
||
func normalizeRelationType(value string) string {
|
||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||
case cpdomain.RelationTypeCP:
|
||
return cpdomain.RelationTypeCP
|
||
case cpdomain.RelationTypeBrother:
|
||
return cpdomain.RelationTypeBrother
|
||
case cpdomain.RelationTypeSister:
|
||
return cpdomain.RelationTypeSister
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
func orderedPair(a int64, b int64) (int64, int64) {
|
||
if a < b {
|
||
return a, b
|
||
}
|
||
return b, a
|
||
}
|
||
|
||
func relationshipIDForPair(userAID int64, userBID int64) string {
|
||
// 关系 ID 固定由无序 pair 派生，配合 active 唯一约束保证 A/B 不会同时有多种 active 关系。
|
||
return fmt.Sprintf("cp_rel:%d:%d", userAID, userBID)
|
||
}
|
||
|
||
func giftSnapshotFromEvent(event cpdomain.GiftEvent) cpdomain.GiftSnapshot {
|
||
return cpdomain.GiftSnapshot{
|
||
GiftID: strings.TrimSpace(event.GiftID),
|
||
GiftName: strings.TrimSpace(event.GiftName),
|
||
GiftIconURL: strings.TrimSpace(event.GiftIconURL),
|
||
GiftAnimationURL: strings.TrimSpace(event.GiftAnimationURL),
|
||
GiftCount: event.GiftCount,
|
||
GiftValue: event.GiftValue,
|
||
BillingReceiptID: strings.TrimSpace(event.BillingReceiptID),
|
||
}
|
||
}
|
||
|
||
func clampPageSize(pageSize int32) int32 {
|
||
if pageSize <= 0 {
|
||
return 20
|
||
}
|
||
if pageSize > 100 {
|
||
return 100
|
||
}
|
||
return pageSize
|
||
}
|
||
|
||
func offset(page int32, pageSize int32) int32 {
|
||
if page <= 1 {
|
||
return 0
|
||
}
|
||
return (page - 1) * clampPageSize(pageSize)
|
||
}
|