218 lines
8.4 KiB
Go
218 lines
8.4 KiB
Go
// 本包承载 user-service 内 Host/Agency/BD 的 MySQL 持久化。
|
||
// 父级 mysql.Repository 负责连接池；本包负责领域行锁、命令幂等记录和事件箱写入。
|
||
package host
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
|
||
mysqldriver "github.com/go-sql-driver/mysql"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
hostdomain "hyapp/services/user-service/internal/domain/host"
|
||
userdomain "hyapp/services/user-service/internal/domain/user"
|
||
)
|
||
|
||
// 存储对象是 Host 领域在 user-service MySQL 上的持久化实现。
|
||
type Repository struct {
|
||
// 数据库连接由父级 user-service MySQL 存储对象持有，本包只复用连接池。
|
||
db *sql.DB
|
||
}
|
||
|
||
// 基于 user-service 共享连接池创建 Host 领域存储对象。
|
||
func New(db *sql.DB) *Repository {
|
||
return &Repository{db: db}
|
||
}
|
||
|
||
const (
|
||
coinSellerProfileColumns = `
|
||
user_id, status, merchant_asset_type,
|
||
created_by_user_id, created_at_ms, updated_at_ms`
|
||
)
|
||
|
||
var (
|
||
// 列清单集中维护，保证查询函数和结果回放使用同一组投影字段。
|
||
// 可空外键在存储边界统一转成 0，让领域结构体保持普通值类型。
|
||
// 角色表不再保存区域；领域读取统一从对应 users 行投影当前区域。
|
||
hostProfileColumns = fmt.Sprintf(`
|
||
user_id, status, %s, COALESCE(current_agency_id, 0),
|
||
COALESCE(current_membership_id, 0),
|
||
COALESCE((
|
||
SELECT owner_user_id
|
||
FROM agencies
|
||
WHERE agencies.app_code = host_profiles.app_code
|
||
AND agencies.agency_id = host_profiles.current_agency_id
|
||
AND agencies.status = 'active'
|
||
LIMIT 1
|
||
), 0),
|
||
source, first_became_host_at_ms, created_at_ms, updated_at_ms`, userRegionProjection("host_profiles", "user_id"))
|
||
agencyColumns = fmt.Sprintf(`
|
||
agency_id, owner_user_id, %s, parent_bd_user_id, name, status,
|
||
join_enabled, max_hosts, created_by_user_id, created_at_ms, updated_at_ms`, userRegionProjection("agencies", "owner_user_id"))
|
||
bdProfileColumns = fmt.Sprintf(`
|
||
user_id, role, %s, COALESCE(parent_leader_user_id, 0), status,
|
||
created_by_user_id, created_at_ms, updated_at_ms`, userRegionProjection("bd_profiles", "user_id"))
|
||
bdLeaderProfileColumns = fmt.Sprintf(`
|
||
user_id, 'bd_leader', %s, 0, status,
|
||
created_by_user_id, created_at_ms, updated_at_ms`, userRegionProjection("bd_leader_profiles", "user_id"))
|
||
managerProfileColumns = fmt.Sprintf(`
|
||
user_id, %s, status, contact_info, created_by_admin_id, created_at_ms, updated_at_ms`, userRegionProjection("manager_profiles", "user_id"))
|
||
agencyMembershipColumns = fmt.Sprintf(`
|
||
membership_id, agency_id, host_user_id, %s, membership_type, status,
|
||
joined_at_ms, COALESCE(ended_at_ms, 0), COALESCE(ended_by_user_id, 0),
|
||
ended_reason, created_at_ms, updated_at_ms`, userRegionProjection("agency_memberships", "host_user_id"))
|
||
agencyApplicationColumns = fmt.Sprintf(`
|
||
application_id, command_id, applicant_user_id, agency_id, %s, status,
|
||
COALESCE(reviewed_by_user_id, 0), review_reason, COALESCE(reviewed_at_ms, 0),
|
||
created_at_ms, updated_at_ms`, userRegionProjection("agency_applications", "applicant_user_id"))
|
||
roleInvitationColumns = fmt.Sprintf(`
|
||
invitation_id, command_id, invitation_type, status, inviter_user_id,
|
||
inviter_bd_user_id, target_user_id, %s, agency_name, parent_bd_user_id,
|
||
COALESCE(parent_leader_user_id, 0), COALESCE(processed_by_user_id, 0),
|
||
process_reason, COALESCE(processed_at_ms, 0), created_at_ms, updated_at_ms`, userRegionProjection("role_invitations", "target_user_id"))
|
||
)
|
||
|
||
type sqlQueryer interface {
|
||
QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
|
||
QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
|
||
}
|
||
|
||
type rowScanner interface {
|
||
Scan(dest ...any) error
|
||
}
|
||
|
||
// nullableRegionID 把领域层“0 表示缺失”的约定转换为 SQL NULL。
|
||
// 这个函数也用于可选用户 ID、成员关系 ID，不只用于区域 ID。
|
||
func nullableRegionID(value int64) any {
|
||
if value <= 0 {
|
||
return nil
|
||
}
|
||
return value
|
||
}
|
||
|
||
func userRegionProjection(tableName string, userIDColumn string) string {
|
||
return fmt.Sprintf(`COALESCE((
|
||
SELECT u.region_id
|
||
FROM users u
|
||
WHERE u.app_code = %s.app_code
|
||
AND u.user_id = %s.%s
|
||
LIMIT 1
|
||
), 0)`, tableName, tableName, userIDColumn)
|
||
}
|
||
|
||
func userRegionValue(ctx context.Context, q sqlQueryer, userID int64, lockClause string) (int64, error) {
|
||
// 只读取 users 当前区域，不校验用户状态；用于历史邀请的发起人区域对齐，
|
||
// 避免旧库残留的邀请区域快照继续参与判断，又不额外改变“邀请已发出后可处理”的状态语义。
|
||
query := "SELECT COALESCE(region_id, 0) FROM users WHERE app_code = ? AND user_id = ?"
|
||
if strings.TrimSpace(lockClause) != "" {
|
||
query += " " + lockClause
|
||
}
|
||
var regionID int64
|
||
err := q.QueryRowContext(ctx, query, appcode.FromContext(ctx), userID).Scan(®ionID)
|
||
if err == sql.ErrNoRows {
|
||
return 0, xerr.New(xerr.NotFound, "user not found")
|
||
}
|
||
return regionID, err
|
||
}
|
||
|
||
func currentUserRegion(ctx context.Context, q sqlQueryer, userID int64, lockClause string) (int64, error) {
|
||
// Host 领域不拥有 users 表，但所有角色流转都依赖用户当前有效区域。
|
||
// 调用方需要串行化用户身份变化时传入 FOR UPDATE 来锁住 users 行。
|
||
query := "SELECT COALESCE(region_id, 0), status FROM users WHERE app_code = ? AND user_id = ?"
|
||
if strings.TrimSpace(lockClause) != "" {
|
||
query += " " + lockClause
|
||
}
|
||
var regionID int64
|
||
var status string
|
||
err := q.QueryRowContext(ctx, query, appcode.FromContext(ctx), userID).Scan(®ionID, &status)
|
||
if err == sql.ErrNoRows {
|
||
return 0, xerr.New(xerr.NotFound, "user not found")
|
||
}
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
if status != "active" {
|
||
return 0, xerr.New(xerr.UserDisabled, "user is not active")
|
||
}
|
||
return regionID, nil
|
||
}
|
||
|
||
func (r *Repository) userRegion(ctx context.Context, q sqlQueryer, userID int64, lockClause string) (int64, error) {
|
||
return currentUserRegion(ctx, q, userID, lockClause)
|
||
}
|
||
|
||
func (r *Repository) userCountry(ctx context.Context, q sqlQueryer, userID int64, lockClause string) (string, error) {
|
||
// App 侧 Agency 推荐按用户当前国家收敛；国家来自用户资料事实，不能由客户端传入。
|
||
query := "SELECT COALESCE(country, ''), status FROM users WHERE app_code = ? AND user_id = ?"
|
||
if strings.TrimSpace(lockClause) != "" {
|
||
query += " " + lockClause
|
||
}
|
||
var country string
|
||
var status string
|
||
err := q.QueryRowContext(ctx, query, appcode.FromContext(ctx), userID).Scan(&country, &status)
|
||
if err == sql.ErrNoRows {
|
||
return "", xerr.New(xerr.NotFound, "user not found")
|
||
}
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if status != "active" {
|
||
return "", xerr.New(xerr.UserDisabled, "user is not active")
|
||
}
|
||
return strings.TrimSpace(country), nil
|
||
}
|
||
|
||
func requireActiveRegion(ctx context.Context, q sqlQueryer, regionID int64) error {
|
||
if regionID <= 0 {
|
||
return xerr.New(xerr.InvalidArgument, "region_id is required")
|
||
}
|
||
var status string
|
||
err := q.QueryRowContext(ctx, `
|
||
SELECT status
|
||
FROM regions
|
||
WHERE app_code = ? AND region_id = ?
|
||
`, appcode.FromContext(ctx), regionID).Scan(&status)
|
||
if err == sql.ErrNoRows {
|
||
return xerr.New(xerr.RegionNotFound, "region not found")
|
||
}
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if status != userdomain.RegionStatusActive {
|
||
return xerr.New(xerr.RegionDisabled, "region is disabled")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func appScopedClause(ctx context.Context, clause string, args ...any) (string, []any) {
|
||
trimmed := strings.TrimSpace(clause)
|
||
if strings.HasPrefix(strings.ToUpper(trimmed), "WHERE ") {
|
||
return "WHERE app_code = ? AND " + strings.TrimSpace(trimmed[len("WHERE "):]), append([]any{appcode.FromContext(ctx)}, args...)
|
||
}
|
||
if trimmed == "" {
|
||
return "WHERE app_code = ?", []any{appcode.FromContext(ctx)}
|
||
}
|
||
return clause, args
|
||
}
|
||
|
||
func mapHostDuplicateError(err error) error {
|
||
// 唯一键是显式行锁之后的最终并发兜底。
|
||
// 这里统一转成领域冲突，避免业务层/传输层感知 MySQL 错误码。
|
||
var mysqlErr *mysqldriver.MySQLError
|
||
if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
|
||
return xerr.New(xerr.Conflict, "host command or relationship already exists")
|
||
}
|
||
return err
|
||
}
|
||
|
||
func requireCommandResult(result hostdomain.CommandResult, commandType string, resultType string) error {
|
||
if result.CommandType != commandType || result.ResultType != resultType {
|
||
// 同一个 command_id 不能跨动作复用，否则调用方无法判断重试语义。
|
||
return xerr.New(xerr.Conflict, "command_id was used by another host command")
|
||
}
|
||
return nil
|
||
}
|