shanshanzhong
ef64a876cd
All checks were successful
Build docker and publish / build (20.15.1) (push) Successful in 5m10s
添加请求追踪中间件以记录请求和响应内容 在用户订阅查询中新增includeExpired参数支持查询历史记录 完善配置系统以支持float64类型默认值解析
222 lines
6.8 KiB
Go
222 lines
6.8 KiB
Go
package user
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"time"
|
||
|
||
"github.com/perfect-panel/server/pkg/constant"
|
||
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
func (m *defaultUserModel) UpdateUserSubscribeCache(ctx context.Context, data *Subscribe) error {
|
||
return m.ClearSubscribeCacheByModels(ctx, data)
|
||
}
|
||
|
||
// QueryActiveSubscriptions returns the number of active subscriptions.
|
||
func (m *defaultUserModel) QueryActiveSubscriptions(ctx context.Context, subscribeId ...int64) (map[int64]int64, error) {
|
||
type SubscriptionCount struct {
|
||
SubscribeId int64
|
||
Total int64
|
||
}
|
||
var result []SubscriptionCount
|
||
err := m.QueryNoCacheCtx(ctx, &result, func(conn *gorm.DB, v interface{}) error {
|
||
return conn.Model(&Subscribe{}).
|
||
Where("subscribe_id IN ? AND `status` IN ?", subscribeId, []int64{1, 0}).
|
||
Select("subscribe_id, COUNT(id) as total").
|
||
Group("subscribe_id").
|
||
Scan(&result).
|
||
Error
|
||
})
|
||
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
resultMap := make(map[int64]int64)
|
||
for _, item := range result {
|
||
resultMap[item.SubscribeId] = item.Total
|
||
}
|
||
|
||
return resultMap, nil
|
||
}
|
||
|
||
func (m *defaultUserModel) FindOneSubscribeByOrderId(ctx context.Context, orderId int64) (*Subscribe, error) {
|
||
var data Subscribe
|
||
err := m.QueryNoCacheCtx(ctx, &data, func(conn *gorm.DB, v interface{}) error {
|
||
return conn.Model(&Subscribe{}).Where("order_id = ?", orderId).First(&data).Error
|
||
})
|
||
return &data, err
|
||
}
|
||
|
||
func (m *defaultUserModel) FindOneSubscribe(ctx context.Context, id int64) (*Subscribe, error) {
|
||
var data Subscribe
|
||
key := fmt.Sprintf("%s%d", cacheUserSubscribeIdPrefix, id)
|
||
err := m.QueryCtx(ctx, &data, key, func(conn *gorm.DB, v interface{}) error {
|
||
return conn.Model(&Subscribe{}).Where("id = ?", id).First(&data).Error
|
||
})
|
||
return &data, err
|
||
}
|
||
|
||
func (m *defaultUserModel) FindUsersSubscribeBySubscribeId(ctx context.Context, subscribeId int64) ([]*Subscribe, error) {
|
||
var data []*Subscribe
|
||
err := m.QueryNoCacheCtx(ctx, &data, func(conn *gorm.DB, v interface{}) error {
|
||
err := conn.Model(&Subscribe{}).Where("subscribe_id = ? AND `status` IN ?", subscribeId, []int64{1, 0}).Find(v).Error
|
||
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// update user subscribe status
|
||
return conn.Model(&Subscribe{}).Where("subscribe_id = ? AND `status` = ?", subscribeId, 0).Update("status", 1).Error
|
||
})
|
||
return data, err
|
||
}
|
||
|
||
// QueryUserSubscribe returns a list of records that meet the conditions.
|
||
func (m *defaultUserModel) QueryUserSubscribe(ctx context.Context, userId int64, status ...int64) ([]*SubscribeDetails, error) {
|
||
var list []*SubscribeDetails
|
||
key := fmt.Sprintf("%s%d", cacheUserSubscribeUserPrefix, userId)
|
||
|
||
// 1. Get includeExpired from Context
|
||
includeExpired := ""
|
||
if v := ctx.Value(constant.CtxKeyIncludeExpired); v != nil {
|
||
includeExpired, _ = v.(string)
|
||
}
|
||
|
||
// 2. If query mode is different, must modify Cache Key
|
||
if includeExpired == "all" {
|
||
key += ":all"
|
||
}
|
||
|
||
err := m.QueryCtx(ctx, &list, key, func(conn *gorm.DB, v interface{}) error {
|
||
// Base condition
|
||
db := conn.Model(&Subscribe{}).Where("`user_id` = ?", userId)
|
||
if len(status) > 0 {
|
||
db = db.Where("`status` IN ?", status)
|
||
}
|
||
|
||
// 3. Adjust SQL based on param
|
||
if includeExpired == "all" {
|
||
// Mode A: Query all history
|
||
return db.Order("created_at DESC").Preload("Subscribe").Find(&list).Error
|
||
}
|
||
|
||
// Mode B: Default only query valid subscriptions
|
||
// Logic: ExpireTime > Now OR FinishedAt >= 7 days ago OR ExpireTime = 0 (Never expire)
|
||
now := time.Now()
|
||
sevenDaysAgo := now.Add(-7 * 24 * time.Hour)
|
||
return db.Where("`expire_time` > ? OR `finished_at` >= ? OR `expire_time` = ?", now, sevenDaysAgo, time.UnixMilli(0)).
|
||
Preload("Subscribe").
|
||
Find(&list).Error
|
||
})
|
||
return list, err
|
||
}
|
||
|
||
// FindOneUserSubscribe finds a subscribeDetails by id.
|
||
func (m *defaultUserModel) FindOneUserSubscribe(ctx context.Context, id int64) (subscribeDetails *SubscribeDetails, err error) {
|
||
//TODO cache
|
||
//key := fmt.Sprintf("%s%d", cacheUserSubscribeUserPrefix, userId)
|
||
subscribeDetails = &SubscribeDetails{} // 初始化结构体，避免nil指针
|
||
err = m.QueryNoCacheCtx(ctx, subscribeDetails, func(conn *gorm.DB, v interface{}) error {
|
||
return conn.Model(&Subscribe{}).Preload("Subscribe").Where("id = ?", id).First(subscribeDetails).Error
|
||
})
|
||
return
|
||
}
|
||
|
||
// FindOneSubscribeByToken finds a record by token.
|
||
func (m *defaultUserModel) FindOneSubscribeByToken(ctx context.Context, token string) (*Subscribe, error) {
|
||
var data Subscribe
|
||
key := fmt.Sprintf("%s%s", cacheUserSubscribeTokenPrefix, token)
|
||
err := m.QueryCtx(ctx, &data, key, func(conn *gorm.DB, v interface{}) error {
|
||
return conn.Model(&Subscribe{}).Where("token = ?", token).First(&data).Error
|
||
})
|
||
return &data, err
|
||
}
|
||
|
||
// UpdateSubscribe updates a record.
|
||
func (m *defaultUserModel) UpdateSubscribe(ctx context.Context, data *Subscribe, tx ...*gorm.DB) error {
|
||
old, err := m.FindOneSubscribe(ctx, data.Id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 使用 defer 确保更新后清理缓存
|
||
defer func() {
|
||
if clearErr := m.ClearSubscribeCacheByModels(ctx, old, data); clearErr != nil {
|
||
// 记录清理缓存错误
|
||
}
|
||
}()
|
||
|
||
return m.ExecNoCacheCtx(ctx, func(conn *gorm.DB) error {
|
||
if len(tx) > 0 {
|
||
conn = tx[0]
|
||
}
|
||
return conn.Model(&Subscribe{}).Where("id = ?", data.Id).Save(data).Error
|
||
})
|
||
}
|
||
|
||
// DeleteSubscribe deletes a record.
|
||
func (m *defaultUserModel) DeleteSubscribe(ctx context.Context, token string, tx ...*gorm.DB) error {
|
||
data, err := m.FindOneSubscribeByToken(ctx, token)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 使用 defer 确保删除后清理缓存
|
||
defer func() {
|
||
if clearErr := m.ClearSubscribeCacheByModels(ctx, data); clearErr != nil {
|
||
// 记录清理缓存错误
|
||
}
|
||
}()
|
||
|
||
return m.ExecNoCacheCtx(ctx, func(conn *gorm.DB) error {
|
||
if len(tx) > 0 {
|
||
conn = tx[0]
|
||
}
|
||
return conn.Where("token = ?", token).Delete(&Subscribe{}).Error
|
||
})
|
||
}
|
||
|
||
// InsertSubscribe insert Subscribe into the database.
|
||
func (m *defaultUserModel) InsertSubscribe(ctx context.Context, data *Subscribe, tx ...*gorm.DB) error {
|
||
// 使用 defer 确保插入后清理相关缓存
|
||
defer func() {
|
||
if clearErr := m.ClearSubscribeCacheByModels(ctx, data); clearErr != nil {
|
||
// 记录清理缓存错误
|
||
}
|
||
}()
|
||
|
||
return m.ExecNoCacheCtx(ctx, func(conn *gorm.DB) error {
|
||
if len(tx) > 0 {
|
||
conn = tx[0]
|
||
}
|
||
return conn.Create(data).Error
|
||
})
|
||
}
|
||
|
||
func (m *defaultUserModel) DeleteSubscribeById(ctx context.Context, id int64, tx ...*gorm.DB) error {
|
||
data, err := m.FindOneSubscribe(ctx, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 使用 defer 确保删除后清理缓存
|
||
defer func() {
|
||
if clearErr := m.ClearSubscribeCacheByModels(ctx, data); clearErr != nil {
|
||
// 记录清理缓存错误
|
||
}
|
||
}()
|
||
|
||
return m.ExecNoCacheCtx(ctx, func(conn *gorm.DB) error {
|
||
if len(tx) > 0 {
|
||
conn = tx[0]
|
||
}
|
||
return conn.Where("id = ?", id).Delete(&Subscribe{}).Error
|
||
})
|
||
}
|
||
|
||
func (m *defaultUserModel) ClearSubscribeCache(ctx context.Context, data ...*Subscribe) error {
|
||
return m.ClearSubscribeCacheByModels(ctx, data...)
|
||
}
|