Files
backend/internal/repository/user_repo.go
lan 90c57f1a1c
All checks were successful
Build Backend / build (push) Successful in 2m7s
Build Backend / build-docker (push) Successful in 1m51s
feat(core): optimize performance and reliability through batching and redis-backed unread counts
This commit introduces several significant architectural improvements to enhance system performance, scalability, and reliability:

- **Performance Optimization (N+1 Problem Resolution)**: Implemented batching mechanisms across `MessageHandler`, `ChatService`, `GroupService`, `MessageService`, and `UserService`. This replaces multiple individual database/cache queries with single batch operations for fetching unread counts, last messages, participants, and user information.
- **Enhanced Unread Count Management**: Migrated unread count tracking to a Redis Hash-based approach (`unread#️⃣{userID}`). This allows for atomic increments/decrements and efficient retrieval of both individual conversation unread counts and total unread counts for a user.
- **Improved Message Sequencing**: Integrated Redis-based sequence (`seq`) pre-allocation for messages to ensure strict ordering and reduce database contention during high-concurrency message creation.
- **Push Service Reliability**: Refactored the push notification system to include persistent `PushRecord` tracking. Added a recovery mechanism (`recoverPendingPushes`) to reload pending notifications from the database upon service startup, ensuring better delivery guarantees.
- **WebSocket Reliability**: Updated the WebSocket hub to move away from in-memory history replay in favor of a client-driven synchronization model (`sync_required` event and `ack` handling), reducing memory overhead and improving connection stability.
- **Cache Layer Enhancements**: Added `HIncrBy` and `IncrBySeq` to the `Cache` interface and its implementations (`RedisCache`, `LayeredCache`) to support the new unread and sequence management logic.
2026-05-04 18:31:03 +08:00

631 lines
20 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package repository
import (
"context"
"strings"
"time"
"with_you/internal/model"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
// UserRepository 用户仓储接口
type UserRepository interface {
Create(user *model.User) error
GetByID(id string) (*model.User, error)
GetUsersByIDs(ctx context.Context, ids []string) (map[string]*model.User, error)
GetByUsername(username string) (*model.User, error)
GetByEmail(email string) (*model.User, error)
GetByPhone(phone string) (*model.User, error)
Update(user *model.User) error
Delete(id string) error
List(page, pageSize int) ([]*model.User, int64, error)
GetFollowers(userID string, page, pageSize int) ([]*model.User, int64, error)
GetFollowing(userID string, page, pageSize int) ([]*model.User, int64, error)
CreateFollow(follow *model.Follow) error
DeleteFollow(followerID, followingID string) error
IsFollowing(followerID, followingID string) (bool, error)
IsFollowingBatch(followerID string, targetUserIDs []string) (map[string]bool, error)
IncrementFollowersCount(userID string) error
DecrementFollowersCount(userID string) error
IncrementFollowingCount(userID string) error
DecrementFollowingCount(userID string) error
RefreshFollowersCount(userID string) error
GetPostsCount(userID string) (int64, error)
GetPostsCountBatch(userIDs []string) (map[string]int64, error)
RefreshFollowingCount(userID string) error
IsBlocked(blockerID, blockedID string) (bool, error)
IsBlockedBatch(blockerID string, blockedIDs []string) (map[string]bool, error)
IsBlockedEitherDirection(userA, userB string) (bool, error)
BlockUserAndCleanupRelations(blockerID, blockedID string) error
UnblockUser(blockerID, blockedID string) error
GetBlockedUsers(blockerID string, page, pageSize int) ([]*model.User, int64, error)
Search(keyword string, page, pageSize int) ([]*model.User, int64, error)
GetMutualFollowStatus(currentUserID string, targetUserIDs []string) (map[string][2]bool, error)
AdminList(page, pageSize int, keyword, status string) ([]*model.User, int64, error)
GetCommentsCount(userID string) (int64, error)
UpdateStatus(userID string, status model.UserStatus) error
// 隐私设置
UpdatePrivacySettings(userID string, settings *model.PrivacySettings) error
GetPrivacySettings(userID string) (*model.PrivacySettings, error)
// 账号注销
RequestDeletion(userID string) error
CancelDeletion(userID string) error
GetPendingDeletionUsers(beforeTime time.Time) ([]*model.User, error)
HardDeleteUser(userID string) error
}
// userRepository 用户仓储实现
type userRepository struct {
db *gorm.DB
}
// NewUserRepository 创建用户仓储
func NewUserRepository(db *gorm.DB) UserRepository {
return &userRepository{db: db}
}
// Create 创建用户
func (r *userRepository) Create(user *model.User) error {
return r.db.Create(user).Error
}
// GetByID 根据ID获取用户
func (r *userRepository) GetByID(id string) (*model.User, error) {
var user model.User
err := r.db.First(&user, "id = ?", id).Error
if err != nil {
return nil, err
}
return &user, nil
}
// GetByUsername 根据用户名获取用户
func (r *userRepository) GetByUsername(username string) (*model.User, error) {
var user model.User
err := r.db.First(&user, "username = ?", username).Error
if err != nil {
return nil, err
}
return &user, nil
}
// GetByEmail 根据邮箱获取用户
func (r *userRepository) GetByEmail(email string) (*model.User, error) {
var user model.User
err := r.db.First(&user, "email = ?", email).Error
if err != nil {
return nil, err
}
return &user, nil
}
// GetByPhone 根据手机号获取用户
func (r *userRepository) GetByPhone(phone string) (*model.User, error) {
var user model.User
err := r.db.First(&user, "phone = ?", phone).Error
if err != nil {
return nil, err
}
return &user, nil
}
// Update 更新用户
func (r *userRepository) Update(user *model.User) error {
return r.db.Save(user).Error
}
// Delete 删除用户
func (r *userRepository) Delete(id string) error {
return r.db.Delete(&model.User{}, "id = ?", id).Error
}
// List 分页获取用户列表
func (r *userRepository) List(page, pageSize int) ([]*model.User, int64, error) {
var users []*model.User
var total int64
r.db.Model(&model.User{}).Count(&total)
offset := (page - 1) * pageSize
err := r.db.Order("created_at DESC, id DESC").Offset(offset).Limit(pageSize).Find(&users).Error
return users, total, err
}
// GetFollowers 获取用户粉丝
func (r *userRepository) GetFollowers(userID string, page, pageSize int) ([]*model.User, int64, error) {
var users []*model.User
var total int64
subQuery := r.db.Model(&model.Follow{}).Where("following_id = ?", userID).Select("follower_id")
r.db.Model(&model.User{}).Where("id IN (?)", subQuery).Count(&total)
offset := (page - 1) * pageSize
err := r.db.Where("id IN (?)", subQuery).
Order("created_at DESC, id DESC").
Offset(offset).Limit(pageSize).
Find(&users).Error
return users, total, err
}
// GetFollowing 获取用户关注
func (r *userRepository) GetFollowing(userID string, page, pageSize int) ([]*model.User, int64, error) {
var users []*model.User
var total int64
subQuery := r.db.Model(&model.Follow{}).Where("follower_id = ?", userID).Select("following_id")
r.db.Model(&model.User{}).Where("id IN (?)", subQuery).Count(&total)
offset := (page - 1) * pageSize
err := r.db.Where("id IN (?)", subQuery).
Order("created_at DESC, id DESC").
Offset(offset).Limit(pageSize).
Find(&users).Error
return users, total, err
}
// CreateFollow 创建关注关系
func (r *userRepository) CreateFollow(follow *model.Follow) error {
return r.db.Create(follow).Error
}
// DeleteFollow 删除关注关系
func (r *userRepository) DeleteFollow(followerID, followingID string) error {
return r.db.Where("follower_id = ? AND following_id = ?", followerID, followingID).Delete(&model.Follow{}).Error
}
// IsFollowing 检查是否关注了某用户
func (r *userRepository) IsFollowing(followerID, followingID string) (bool, error) {
var count int64
err := r.db.Model(&model.Follow{}).Where("follower_id = ? AND following_id = ?", followerID, followingID).Count(&count).Error
if err != nil {
return false, err
}
return count > 0, nil
}
// IsFollowingBatch 批量检查当前用户是否关注了目标用户(解决 N+1 问题)
// 返回 map[targetUserID]bool
func (r *userRepository) IsFollowingBatch(followerID string, targetUserIDs []string) (map[string]bool, error) {
result := make(map[string]bool)
if len(targetUserIDs) == 0 || followerID == "" {
return result, nil
}
// 初始化所有目标用户为 false
for _, id := range targetUserIDs {
result[id] = false
}
// 一次性查询所有关注记录
var followingIDs []string
err := r.db.Model(&model.Follow{}).
Where("follower_id = ? AND following_id IN ?", followerID, targetUserIDs).
Pluck("following_id", &followingIDs).Error
if err != nil {
return nil, err
}
// 更新已关注的用户
for _, id := range followingIDs {
result[id] = true
}
return result, nil
}
// IncrementFollowersCount 增加用户粉丝数
func (r *userRepository) IncrementFollowersCount(userID string) error {
return r.db.Model(&model.User{}).Where("id = ?", userID).
UpdateColumn("followers_count", gorm.Expr("followers_count + 1")).Error
}
// DecrementFollowersCount 减少用户粉丝数
func (r *userRepository) DecrementFollowersCount(userID string) error {
return r.db.Model(&model.User{}).Where("id = ? AND followers_count > 0", userID).
UpdateColumn("followers_count", gorm.Expr("followers_count - 1")).Error
}
// IncrementFollowingCount 增加用户关注数
func (r *userRepository) IncrementFollowingCount(userID string) error {
return r.db.Model(&model.User{}).Where("id = ?", userID).
UpdateColumn("following_count", gorm.Expr("following_count + 1")).Error
}
// DecrementFollowingCount 减少用户关注数
func (r *userRepository) DecrementFollowingCount(userID string) error {
return r.db.Model(&model.User{}).Where("id = ? AND following_count > 0", userID).
UpdateColumn("following_count", gorm.Expr("following_count - 1")).Error
}
// RefreshFollowersCount 刷新用户粉丝数(通过实际计数)
func (r *userRepository) RefreshFollowersCount(userID string) error {
var count int64
err := r.db.Model(&model.Follow{}).Where("following_id = ?", userID).Count(&count).Error
if err != nil {
return err
}
return r.db.Model(&model.User{}).Where("id = ?", userID).
UpdateColumn("followers_count", count).Error
}
// GetPostsCount 获取用户帖子数(实时计算)
func (r *userRepository) GetPostsCount(userID string) (int64, error) {
var count int64
err := r.db.Model(&model.Post{}).
Where("user_id = ? AND status = ?", userID, model.PostStatusPublished).
Count(&count).Error
return count, err
}
// GetPostsCountBatch 批量获取用户帖子数(实时计算)
// 返回 map[userID]postsCount
func (r *userRepository) GetPostsCountBatch(userIDs []string) (map[string]int64, error) {
result := make(map[string]int64)
if len(userIDs) == 0 {
return result, nil
}
// 初始化所有用户ID的计数为0
for _, userID := range userIDs {
result[userID] = 0
}
// 使用 GROUP BY 一次性查询所有用户的帖子数
type CountResult struct {
UserID string
Count int64
}
var counts []CountResult
err := r.db.Model(&model.Post{}).
Select("user_id, count(*) as count").
Where("user_id IN ? AND status = ?", userIDs, model.PostStatusPublished).
Group("user_id").
Scan(&counts).Error
if err != nil {
return nil, err
}
// 更新查询结果
for _, c := range counts {
result[c.UserID] = c.Count
}
return result, nil
}
// RefreshFollowingCount 刷新用户关注数(通过实际计数)
func (r *userRepository) RefreshFollowingCount(userID string) error {
var count int64
err := r.db.Model(&model.Follow{}).Where("follower_id = ?", userID).Count(&count).Error
if err != nil {
return err
}
return r.db.Model(&model.User{}).Where("id = ?", userID).
UpdateColumn("following_count", count).Error
}
// IsBlocked 检查拉黑关系是否存在blocker -> blocked
func (r *userRepository) IsBlocked(blockerID, blockedID string) (bool, error) {
var count int64
err := r.db.Model(&model.UserBlock{}).
Where("blocker_id = ? AND blocked_id = ?", blockerID, blockedID).
Count(&count).Error
if err != nil {
return false, err
}
return count > 0, nil
}
// IsBlockedBatch 批量检查拉黑关系(解决 N+1 问题)
// 返回 map[blockedUserID]bool
func (r *userRepository) IsBlockedBatch(blockerID string, blockedIDs []string) (map[string]bool, error) {
result := make(map[string]bool)
if len(blockedIDs) == 0 || blockerID == "" {
return result, nil
}
// 初始化所有目标用户为 false
for _, id := range blockedIDs {
result[id] = false
}
// 一次性查询所有拉黑记录
var blockedUserIDs []string
err := r.db.Model(&model.UserBlock{}).
Where("blocker_id = ? AND blocked_id IN ?", blockerID, blockedIDs).
Pluck("blocked_id", &blockedUserIDs).Error
if err != nil {
return nil, err
}
// 更新已拉黑的用户
for _, id := range blockedUserIDs {
result[id] = true
}
return result, nil
}
// IsBlockedEitherDirection 检查是否任一方向存在拉黑
func (r *userRepository) IsBlockedEitherDirection(userA, userB string) (bool, error) {
var count int64
err := r.db.Model(&model.UserBlock{}).
Where("(blocker_id = ? AND blocked_id = ?) OR (blocker_id = ? AND blocked_id = ?)",
userA, userB, userB, userA).
Count(&count).Error
if err != nil {
return false, err
}
return count > 0, nil
}
// BlockUserAndCleanupRelations 拉黑用户并清理双向关注关系(事务)
func (r *userRepository) BlockUserAndCleanupRelations(blockerID, blockedID string) error {
return r.db.Transaction(func(tx *gorm.DB) error {
block := &model.UserBlock{
BlockerID: blockerID,
BlockedID: blockedID,
}
if err := tx.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "blocker_id"}, {Name: "blocked_id"}},
DoNothing: true,
}).Create(block).Error; err != nil {
return err
}
if err := tx.Where("follower_id = ? AND following_id = ?", blockerID, blockedID).
Delete(&model.Follow{}).Error; err != nil {
return err
}
if err := tx.Where("follower_id = ? AND following_id = ?", blockedID, blockerID).
Delete(&model.Follow{}).Error; err != nil {
return err
}
for _, uid := range []string{blockerID, blockedID} {
if err := tx.Model(&model.User{}).Where("id = ?", uid).
UpdateColumn("followers_count", tx.Model(&model.Follow{}).
Select("COUNT(*)").Where("following_id = ?", uid)).Error; err != nil {
return err
}
if err := tx.Model(&model.User{}).Where("id = ?", uid).
UpdateColumn("following_count", tx.Model(&model.Follow{}).
Select("COUNT(*)").Where("follower_id = ?", uid)).Error; err != nil {
return err
}
}
return nil
})
}
// UnblockUser 取消拉黑
func (r *userRepository) UnblockUser(blockerID, blockedID string) error {
return r.db.Where("blocker_id = ? AND blocked_id = ?", blockerID, blockedID).
Delete(&model.UserBlock{}).Error
}
// GetBlockedUsers 获取用户黑名单列表
func (r *userRepository) GetBlockedUsers(blockerID string, page, pageSize int) ([]*model.User, int64, error) {
var users []*model.User
var total int64
subQuery := r.db.Model(&model.UserBlock{}).Where("blocker_id = ?", blockerID).Select("blocked_id")
r.db.Model(&model.User{}).Where("id IN (?)", subQuery).Count(&total)
offset := (page - 1) * pageSize
err := r.db.Where("id IN (?)", subQuery).
Order("created_at DESC, id DESC").
Offset(offset).
Limit(pageSize).
Find(&users).Error
return users, total, err
}
// Search 搜索用户
func (r *userRepository) Search(keyword string, page, pageSize int) ([]*model.User, int64, error) {
var users []*model.User
var total int64
query := r.db.Model(&model.User{})
// 搜索用户名、昵称、简介
if keyword != "" {
if r.db.Dialector.Name() == "postgres" {
searchPattern := "%" + keyword + "%"
query = query.Where(
"username ILIKE ? OR nickname ILIKE ? OR bio ILIKE ?",
searchPattern, searchPattern, searchPattern,
)
} else {
// SQLite: 使用 LOWER() 实现大小写不敏感的模糊搜索
lowerSearchPattern := "%" + strings.ToLower(keyword) + "%"
query = query.Where(
"LOWER(username) LIKE ? OR LOWER(nickname) LIKE ? OR LOWER(bio) LIKE ?",
lowerSearchPattern, lowerSearchPattern, lowerSearchPattern,
)
}
}
query.Count(&total)
offset := (page - 1) * pageSize
err := query.Offset(offset).Limit(pageSize).Order("created_at DESC").Find(&users).Error
return users, total, err
}
// GetMutualFollowStatus 批量获取双向关注状态
// 返回 map[userID][isFollowing, isFollowingMe]
func (r *userRepository) GetMutualFollowStatus(currentUserID string, targetUserIDs []string) (map[string][2]bool, error) {
result := make(map[string][2]bool)
if len(targetUserIDs) == 0 {
return result, nil
}
// 初始化所有目标用户为未关注状态
for _, userID := range targetUserIDs {
result[userID] = [2]bool{false, false}
}
// 查询当前用户关注了哪些目标用户 (isFollowing)
var followingIDs []string
err := r.db.Model(&model.Follow{}).
Where("follower_id = ? AND following_id IN ?", currentUserID, targetUserIDs).
Pluck("following_id", &followingIDs).Error
if err != nil {
return nil, err
}
for _, id := range followingIDs {
status := result[id]
status[0] = true
result[id] = status
}
// 查询哪些目标用户关注了当前用户 (isFollowingMe)
var followerIDs []string
err = r.db.Model(&model.Follow{}).
Where("follower_id IN ? AND following_id = ?", targetUserIDs, currentUserID).
Pluck("follower_id", &followerIDs).Error
if err != nil {
return nil, err
}
for _, id := range followerIDs {
status := result[id]
status[1] = true
result[id] = status
}
return result, nil
}
// AdminList 管理端分页获取用户列表(支持关键词和状态筛选)
func (r *userRepository) AdminList(page, pageSize int, keyword, status string) ([]*model.User, int64, error) {
var users []*model.User
var total int64
query := r.db.Model(&model.User{})
// 关键词搜索(用户名、昵称、邮箱)
if keyword != "" {
if r.db.Dialector.Name() == "postgres" {
query = query.Where(
"username ILIKE ? OR nickname ILIKE ? OR email ILIKE ?",
"%"+keyword+"%", "%"+keyword+"%", "%"+keyword+"%",
)
} else {
lowerSearchPattern := "%" + strings.ToLower(keyword) + "%"
query = query.Where(
"LOWER(username) LIKE ? OR LOWER(nickname) LIKE ? OR LOWER(email) LIKE ?",
lowerSearchPattern, lowerSearchPattern, lowerSearchPattern,
)
}
}
// 状态筛选
if status != "" {
query = query.Where("status = ?", status)
}
query.Count(&total)
offset := (page - 1) * pageSize
err := query.Order("created_at DESC, id DESC").Offset(offset).Limit(pageSize).Find(&users).Error
return users, total, err
}
// GetCommentsCount 获取用户评论数
func (r *userRepository) GetCommentsCount(userID string) (int64, error) {
var count int64
err := r.db.Model(&model.Comment{}).
Where("user_id = ?", userID).
Count(&count).Error
return count, err
}
// UpdateStatus 更新用户状态
func (r *userRepository) UpdateStatus(userID string, status model.UserStatus) error {
return r.db.Model(&model.User{}).Where("id = ?", userID).Update("status", status).Error
}
// UpdatePrivacySettings 更新隐私设置
func (r *userRepository) UpdatePrivacySettings(userID string, settings *model.PrivacySettings) error {
return r.db.Model(&model.User{}).Where("id = ?", userID).Updates(map[string]interface{}{
"privacy_followers_visibility": settings.FollowersVisibility,
"privacy_following_visibility": settings.FollowingVisibility,
"privacy_posts_visibility": settings.PostsVisibility,
"privacy_favorites_visibility": settings.FavoritesVisibility,
}).Error
}
// GetPrivacySettings 获取隐私设置
func (r *userRepository) GetPrivacySettings(userID string) (*model.PrivacySettings, error) {
var user model.User
err := r.db.Select("privacy_followers_visibility, privacy_following_visibility, privacy_posts_visibility, privacy_favorites_visibility").
First(&user, "id = ?", userID).Error
if err != nil {
return nil, err
}
return &user.PrivacySettings, nil
}
// RequestDeletion 申请注销账号
func (r *userRepository) RequestDeletion(userID string) error {
now := time.Now()
return r.db.Model(&model.User{}).Where("id = ?", userID).Updates(map[string]interface{}{
"deletion_requested_at": &now,
"status": model.UserStatusPendingDeletion,
}).Error
}
// CancelDeletion 取消注销
func (r *userRepository) CancelDeletion(userID string) error {
return r.db.Model(&model.User{}).Where("id = ?", userID).Updates(map[string]interface{}{
"deletion_requested_at": nil,
"status": model.UserStatusActive,
}).Error
}
// GetPendingDeletionUsers 获取待删除的用户列表(注销申请时间早于指定时间)
func (r *userRepository) GetPendingDeletionUsers(beforeTime time.Time) ([]*model.User, error) {
var users []*model.User
err := r.db.Where("deletion_requested_at IS NOT NULL AND deletion_requested_at < ?", beforeTime).
Find(&users).Error
return users, err
}
// HardDeleteUser 硬删除用户(永久删除)
func (r *userRepository) HardDeleteUser(userID string) error {
return r.db.Unscoped().Delete(&model.User{}, "id = ?", userID).Error
}
// GetUsersByIDs 批量获取用户信息
func (r *userRepository) GetUsersByIDs(ctx context.Context, ids []string) (map[string]*model.User, error) {
result := make(map[string]*model.User, len(ids))
if len(ids) == 0 {
return result, nil
}
var users []*model.User
err := r.db.WithContext(ctx).Where("id IN ?", ids).Find(&users).Error
if err != nil {
return nil, err
}
for _, u := range users {
result[u.ID] = u
}
return result, nil
}