5906f9ab98
## 数据完整性修复 (fix-critical-data-integrity) - 添加 error_translate.go 统一错误转换层 - 修复 nil 输入和 NotFound 错误处理 - 增强仓储层错误一致性 ## 仓储一致性修复 (fix-high-repository-consistency) - Group schema 添加 default_validity_days 字段 - Account schema 添加 proxy edge 关联 - 新增 UsageLog ent schema 定义 - 修复 UpdateBalance/UpdateConcurrency 受影响行数校验 ## 数据卫生修复 (fix-medium-data-hygiene) - UserSubscription 添加软删除支持 (SoftDeleteMixin) - RedeemCode/Setting 添加硬删除策略文档 - account_groups/user_allowed_groups 的 created_at 声明 timestamptz - 停止写入 legacy users.allowed_groups 列 - 新增迁移: 011-014 (索引优化、软删除、孤立数据审计、列清理) ## 测试补充 - 添加 UserSubscription 软删除测试 - 添加迁移回归测试 - 添加 NotFound 错误测试 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
402 lines
11 KiB
Go
402 lines
11 KiB
Go
package repository
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"sort"
|
||
|
||
dbent "github.com/Wei-Shaw/sub2api/ent"
|
||
dbuser "github.com/Wei-Shaw/sub2api/ent/user"
|
||
"github.com/Wei-Shaw/sub2api/ent/userallowedgroup"
|
||
"github.com/Wei-Shaw/sub2api/ent/usersubscription"
|
||
"github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
|
||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||
)
|
||
|
||
type userRepository struct {
|
||
client *dbent.Client
|
||
sql sqlExecutor
|
||
}
|
||
|
||
func NewUserRepository(client *dbent.Client, sqlDB *sql.DB) service.UserRepository {
|
||
return newUserRepositoryWithSQL(client, sqlDB)
|
||
}
|
||
|
||
func newUserRepositoryWithSQL(client *dbent.Client, sqlq sqlExecutor) *userRepository {
|
||
return &userRepository{client: client, sql: sqlq}
|
||
}
|
||
|
||
func (r *userRepository) Create(ctx context.Context, userIn *service.User) error {
|
||
if userIn == nil {
|
||
return nil
|
||
}
|
||
|
||
// 统一使用 ent 的事务：保证用户与允许分组的更新原子化，
|
||
// 并避免基于 *sql.Tx 手动构造 ent client 导致的 ExecQuerier 断言错误。
|
||
tx, err := r.client.Tx(ctx)
|
||
if err != nil && !errors.Is(err, dbent.ErrTxStarted) {
|
||
return err
|
||
}
|
||
|
||
var txClient *dbent.Client
|
||
if err == nil {
|
||
defer func() { _ = tx.Rollback() }()
|
||
txClient = tx.Client()
|
||
} else {
|
||
// 已处于外部事务中（ErrTxStarted），复用当前 client 并由调用方负责提交/回滚。
|
||
txClient = r.client
|
||
}
|
||
|
||
created, err := txClient.User.Create().
|
||
SetEmail(userIn.Email).
|
||
SetUsername(userIn.Username).
|
||
SetWechat(userIn.Wechat).
|
||
SetNotes(userIn.Notes).
|
||
SetPasswordHash(userIn.PasswordHash).
|
||
SetRole(userIn.Role).
|
||
SetBalance(userIn.Balance).
|
||
SetConcurrency(userIn.Concurrency).
|
||
SetStatus(userIn.Status).
|
||
Save(ctx)
|
||
if err != nil {
|
||
return translatePersistenceError(err, nil, service.ErrEmailExists)
|
||
}
|
||
|
||
if err := r.syncUserAllowedGroupsWithClient(ctx, txClient, created.ID, userIn.AllowedGroups); err != nil {
|
||
return err
|
||
}
|
||
|
||
if tx != nil {
|
||
if err := tx.Commit(); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
applyUserEntityToService(userIn, created)
|
||
return nil
|
||
}
|
||
|
||
func (r *userRepository) GetByID(ctx context.Context, id int64) (*service.User, error) {
|
||
m, err := r.client.User.Query().Where(dbuser.IDEQ(id)).Only(ctx)
|
||
if err != nil {
|
||
return nil, translatePersistenceError(err, service.ErrUserNotFound, nil)
|
||
}
|
||
|
||
out := userEntityToService(m)
|
||
groups, err := r.loadAllowedGroups(ctx, []int64{id})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if v, ok := groups[id]; ok {
|
||
out.AllowedGroups = v
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (r *userRepository) GetByEmail(ctx context.Context, email string) (*service.User, error) {
|
||
m, err := r.client.User.Query().Where(dbuser.EmailEQ(email)).Only(ctx)
|
||
if err != nil {
|
||
return nil, translatePersistenceError(err, service.ErrUserNotFound, nil)
|
||
}
|
||
|
||
out := userEntityToService(m)
|
||
groups, err := r.loadAllowedGroups(ctx, []int64{m.ID})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if v, ok := groups[m.ID]; ok {
|
||
out.AllowedGroups = v
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (r *userRepository) Update(ctx context.Context, userIn *service.User) error {
|
||
if userIn == nil {
|
||
return nil
|
||
}
|
||
|
||
// 使用 ent 事务包裹用户更新与 allowed_groups 同步，避免跨层事务不一致。
|
||
tx, err := r.client.Tx(ctx)
|
||
if err != nil && !errors.Is(err, dbent.ErrTxStarted) {
|
||
return err
|
||
}
|
||
|
||
var txClient *dbent.Client
|
||
if err == nil {
|
||
defer func() { _ = tx.Rollback() }()
|
||
txClient = tx.Client()
|
||
} else {
|
||
// 已处于外部事务中（ErrTxStarted），复用当前 client 并由调用方负责提交/回滚。
|
||
txClient = r.client
|
||
}
|
||
|
||
updated, err := txClient.User.UpdateOneID(userIn.ID).
|
||
SetEmail(userIn.Email).
|
||
SetUsername(userIn.Username).
|
||
SetWechat(userIn.Wechat).
|
||
SetNotes(userIn.Notes).
|
||
SetPasswordHash(userIn.PasswordHash).
|
||
SetRole(userIn.Role).
|
||
SetBalance(userIn.Balance).
|
||
SetConcurrency(userIn.Concurrency).
|
||
SetStatus(userIn.Status).
|
||
Save(ctx)
|
||
if err != nil {
|
||
return translatePersistenceError(err, service.ErrUserNotFound, service.ErrEmailExists)
|
||
}
|
||
|
||
if err := r.syncUserAllowedGroupsWithClient(ctx, txClient, updated.ID, userIn.AllowedGroups); err != nil {
|
||
return err
|
||
}
|
||
|
||
if tx != nil {
|
||
if err := tx.Commit(); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
userIn.UpdatedAt = updated.UpdatedAt
|
||
return nil
|
||
}
|
||
|
||
func (r *userRepository) Delete(ctx context.Context, id int64) error {
|
||
affected, err := r.client.User.Delete().Where(dbuser.IDEQ(id)).Exec(ctx)
|
||
if err != nil {
|
||
return translatePersistenceError(err, service.ErrUserNotFound, nil)
|
||
}
|
||
if affected == 0 {
|
||
return service.ErrUserNotFound
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *userRepository) List(ctx context.Context, params pagination.PaginationParams) ([]service.User, *pagination.PaginationResult, error) {
|
||
return r.ListWithFilters(ctx, params, "", "", "")
|
||
}
|
||
|
||
func (r *userRepository) ListWithFilters(ctx context.Context, params pagination.PaginationParams, status, role, search string) ([]service.User, *pagination.PaginationResult, error) {
|
||
q := r.client.User.Query()
|
||
|
||
if status != "" {
|
||
q = q.Where(dbuser.StatusEQ(status))
|
||
}
|
||
if role != "" {
|
||
q = q.Where(dbuser.RoleEQ(role))
|
||
}
|
||
if search != "" {
|
||
q = q.Where(
|
||
dbuser.Or(
|
||
dbuser.EmailContainsFold(search),
|
||
dbuser.UsernameContainsFold(search),
|
||
dbuser.WechatContainsFold(search),
|
||
),
|
||
)
|
||
}
|
||
|
||
total, err := q.Clone().Count(ctx)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
|
||
users, err := q.
|
||
Offset(params.Offset()).
|
||
Limit(params.Limit()).
|
||
Order(dbent.Desc(dbuser.FieldID)).
|
||
All(ctx)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
|
||
outUsers := make([]service.User, 0, len(users))
|
||
if len(users) == 0 {
|
||
return outUsers, paginationResultFromTotal(int64(total), params), nil
|
||
}
|
||
|
||
userIDs := make([]int64, 0, len(users))
|
||
userMap := make(map[int64]*service.User, len(users))
|
||
for i := range users {
|
||
userIDs = append(userIDs, users[i].ID)
|
||
u := userEntityToService(users[i])
|
||
outUsers = append(outUsers, *u)
|
||
userMap[u.ID] = &outUsers[len(outUsers)-1]
|
||
}
|
||
|
||
// Batch load active subscriptions with groups to avoid N+1.
|
||
subs, err := r.client.UserSubscription.Query().
|
||
Where(
|
||
usersubscription.UserIDIn(userIDs...),
|
||
usersubscription.StatusEQ(service.SubscriptionStatusActive),
|
||
).
|
||
WithGroup().
|
||
All(ctx)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
|
||
for i := range subs {
|
||
if u, ok := userMap[subs[i].UserID]; ok {
|
||
u.Subscriptions = append(u.Subscriptions, *userSubscriptionEntityToService(subs[i]))
|
||
}
|
||
}
|
||
|
||
allowedGroupsByUser, err := r.loadAllowedGroups(ctx, userIDs)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
for id, u := range userMap {
|
||
if groups, ok := allowedGroupsByUser[id]; ok {
|
||
u.AllowedGroups = groups
|
||
}
|
||
}
|
||
|
||
return outUsers, paginationResultFromTotal(int64(total), params), nil
|
||
}
|
||
|
||
func (r *userRepository) UpdateBalance(ctx context.Context, id int64, amount float64) error {
|
||
client := clientFromContext(ctx, r.client)
|
||
n, err := client.User.Update().Where(dbuser.IDEQ(id)).AddBalance(amount).Save(ctx)
|
||
if err != nil {
|
||
return translatePersistenceError(err, service.ErrUserNotFound, nil)
|
||
}
|
||
if n == 0 {
|
||
return service.ErrUserNotFound
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *userRepository) DeductBalance(ctx context.Context, id int64, amount float64) error {
|
||
client := clientFromContext(ctx, r.client)
|
||
n, err := client.User.Update().
|
||
Where(dbuser.IDEQ(id), dbuser.BalanceGTE(amount)).
|
||
AddBalance(-amount).
|
||
Save(ctx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if n == 0 {
|
||
return service.ErrInsufficientBalance
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *userRepository) UpdateConcurrency(ctx context.Context, id int64, amount int) error {
|
||
client := clientFromContext(ctx, r.client)
|
||
n, err := client.User.Update().Where(dbuser.IDEQ(id)).AddConcurrency(amount).Save(ctx)
|
||
if err != nil {
|
||
return translatePersistenceError(err, service.ErrUserNotFound, nil)
|
||
}
|
||
if n == 0 {
|
||
return service.ErrUserNotFound
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *userRepository) ExistsByEmail(ctx context.Context, email string) (bool, error) {
|
||
return r.client.User.Query().Where(dbuser.EmailEQ(email)).Exist(ctx)
|
||
}
|
||
|
||
func (r *userRepository) RemoveGroupFromAllowedGroups(ctx context.Context, groupID int64) (int64, error) {
|
||
// 仅操作 user_allowed_groups 联接表，legacy users.allowed_groups 列已弃用。
|
||
affected, err := r.client.UserAllowedGroup.Delete().
|
||
Where(userallowedgroup.GroupIDEQ(groupID)).
|
||
Exec(ctx)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return int64(affected), nil
|
||
}
|
||
|
||
func (r *userRepository) GetFirstAdmin(ctx context.Context) (*service.User, error) {
|
||
m, err := r.client.User.Query().
|
||
Where(
|
||
dbuser.RoleEQ(service.RoleAdmin),
|
||
dbuser.StatusEQ(service.StatusActive),
|
||
).
|
||
Order(dbent.Asc(dbuser.FieldID)).
|
||
First(ctx)
|
||
if err != nil {
|
||
return nil, translatePersistenceError(err, service.ErrUserNotFound, nil)
|
||
}
|
||
|
||
out := userEntityToService(m)
|
||
groups, err := r.loadAllowedGroups(ctx, []int64{m.ID})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if v, ok := groups[m.ID]; ok {
|
||
out.AllowedGroups = v
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (r *userRepository) loadAllowedGroups(ctx context.Context, userIDs []int64) (map[int64][]int64, error) {
|
||
out := make(map[int64][]int64, len(userIDs))
|
||
if len(userIDs) == 0 {
|
||
return out, nil
|
||
}
|
||
|
||
rows, err := r.client.UserAllowedGroup.Query().
|
||
Where(userallowedgroup.UserIDIn(userIDs...)).
|
||
All(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
for i := range rows {
|
||
out[rows[i].UserID] = append(out[rows[i].UserID], rows[i].GroupID)
|
||
}
|
||
|
||
for userID := range out {
|
||
sort.Slice(out[userID], func(i, j int) bool { return out[userID][i] < out[userID][j] })
|
||
}
|
||
|
||
return out, nil
|
||
}
|
||
|
||
// syncUserAllowedGroupsWithClient 在 ent client/事务内同步用户允许分组：
|
||
// 仅操作 user_allowed_groups 联接表，legacy users.allowed_groups 列已弃用。
|
||
func (r *userRepository) syncUserAllowedGroupsWithClient(ctx context.Context, client *dbent.Client, userID int64, groupIDs []int64) error {
|
||
if client == nil {
|
||
return nil
|
||
}
|
||
|
||
// Keep join table as the source of truth for reads.
|
||
if _, err := client.UserAllowedGroup.Delete().Where(userallowedgroup.UserIDEQ(userID)).Exec(ctx); err != nil {
|
||
return err
|
||
}
|
||
|
||
unique := make(map[int64]struct{}, len(groupIDs))
|
||
for _, id := range groupIDs {
|
||
if id <= 0 {
|
||
continue
|
||
}
|
||
unique[id] = struct{}{}
|
||
}
|
||
|
||
if len(unique) > 0 {
|
||
creates := make([]*dbent.UserAllowedGroupCreate, 0, len(unique))
|
||
for groupID := range unique {
|
||
creates = append(creates, client.UserAllowedGroup.Create().SetUserID(userID).SetGroupID(groupID))
|
||
}
|
||
if err := client.UserAllowedGroup.
|
||
CreateBulk(creates...).
|
||
OnConflictColumns(userallowedgroup.FieldUserID, userallowedgroup.FieldGroupID).
|
||
DoNothing().
|
||
Exec(ctx); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func applyUserEntityToService(dst *service.User, src *dbent.User) {
|
||
if dst == nil || src == nil {
|
||
return
|
||
}
|
||
dst.ID = src.ID
|
||
dst.CreatedAt = src.CreatedAt
|
||
dst.UpdatedAt = src.UpdatedAt
|
||
}
|