Files
hi-server/internal/model/subscribe/default.go
T
shanshanzhong147 19d28a8f89 修复(#1): 服务器用户列表缓存按 protocol 隔离 + 兜底不写缓存
服务器用户列表缓存跨协议污染修复(HIF-1 / 详见 PR #5 四件套):

1. 缓存 key 加 protocol 维度(`server:user:{server_id}:{protocol}`),对齐 ServerConfig 已有约定
2. 显式枚举协议清除用户列表缓存(AllProtocols + ServerUserListCacheKeysForServer),不用 SCAN
3. 三个兜底分支不写缓存 + Errorw 日志(带 server_id + protocol 字段)
4. hysteria2 → hysteria 兼容归一化 + 6 个新单测

Closes HIF-1
2026-06-03 05:37:03 -07:00

190 lines
5.3 KiB
Go

package subscribe
import (
"context"
"errors"
"fmt"
"strings"
"github.com/perfect-panel/server/internal/model/node"
"github.com/perfect-panel/server/pkg/cache"
"github.com/perfect-panel/server/pkg/tool"
"github.com/redis/go-redis/v9"
"gorm.io/gorm"
)
var _ Model = (*customSubscribeModel)(nil)
var (
cacheSubscribeIdPrefix = "cache:subscribe:id:"
cacheUserSubscribeUserPrefix = "cache:user:subscribe:user:"
)
type (
Model interface {
subscribeModel
customSubscribeLogicModel
}
subscribeModel interface {
Insert(ctx context.Context, data *Subscribe, tx ...*gorm.DB) error
FindOne(ctx context.Context, id int64) (*Subscribe, error)
Update(ctx context.Context, data *Subscribe, tx ...*gorm.DB) error
Delete(ctx context.Context, id int64, tx ...*gorm.DB) error
Transaction(ctx context.Context, fn func(db *gorm.DB) error) error
}
customSubscribeModel struct {
*defaultSubscribeModel
}
defaultSubscribeModel struct {
cache.CachedConn
table string
}
)
func newSubscribeModel(db *gorm.DB, c *redis.Client) *defaultSubscribeModel {
return &defaultSubscribeModel{
CachedConn: cache.NewConn(db, c),
table: "`subscribe`",
}
}
//nolint:unused
func (m *defaultSubscribeModel) batchGetCacheKeys(Subscribes ...*Subscribe) []string {
var keys []string
for _, subscribe := range Subscribes {
keys = append(keys, m.getCacheKeys(subscribe)...)
}
return keys
}
func (m *defaultSubscribeModel) getCacheKeys(data *Subscribe) []string {
if data == nil {
return []string{}
}
var keys []string
if data.Nodes != "" {
var nodes []*node.Node
ids := strings.Split(data.Nodes, ",")
err := m.QueryNoCacheCtx(context.Background(), &nodes, func(conn *gorm.DB, v interface{}) error {
return conn.Model(&node.Node{}).Where("id IN (?)", tool.StringSliceToInt64Slice(ids)).Find(&nodes).Error
})
if err == nil {
for _, n := range nodes {
keys = append(keys, node.ServerUserListCacheKeysForServer(n.ServerId)...)
}
}
}
if data.NodeTags != "" {
var nodes []*node.Node
tags := tool.RemoveDuplicateElements(strings.Split(data.NodeTags, ",")...)
err := m.QueryNoCacheCtx(context.Background(), &nodes, func(conn *gorm.DB, v interface{}) error {
return conn.Model(&node.Node{}).Scopes(InSet("tags", tags)).Find(&nodes).Error
})
if err == nil {
for _, n := range nodes {
keys = append(keys, node.ServerUserListCacheKeysForServer(n.ServerId)...)
}
}
}
return append(keys, fmt.Sprintf("%s%v", cacheSubscribeIdPrefix, data.Id))
}
func (m *defaultSubscribeModel) Insert(ctx context.Context, data *Subscribe, tx ...*gorm.DB) error {
err := m.ExecCtx(ctx, func(conn *gorm.DB) error {
if len(tx) > 0 {
conn = tx[0]
}
return conn.Create(&data).Error
}, m.getCacheKeys(data)...)
return err
}
func (m *defaultSubscribeModel) FindOne(ctx context.Context, id int64) (*Subscribe, error) {
SubscribeIdKey := fmt.Sprintf("%s%v", cacheSubscribeIdPrefix, id)
var resp Subscribe
err := m.QueryCtx(ctx, &resp, SubscribeIdKey, func(conn *gorm.DB, v interface{}) error {
return conn.Model(&Subscribe{}).Where("`id` = ?", id).First(&resp).Error
})
switch {
case err == nil:
return &resp, nil
default:
return nil, err
}
}
func (m *defaultSubscribeModel) Update(ctx context.Context, data *Subscribe, tx ...*gorm.DB) error {
old, err := m.FindOne(ctx, data.Id)
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
var userIds []int64
err = m.QueryNoCacheCtx(ctx, &userIds, func(conn *gorm.DB, v interface{}) error {
return conn.Table("user_subscribe").
Where("subscribe_id = ? AND status IN (0, 1)", data.Id).
Distinct("user_id").
Pluck("user_id", &userIds).Error
})
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
userSubscribeCacheKeys := make([]string, 0, len(userIds))
for _, userId := range userIds {
userSubscribeCacheKeys = append(userSubscribeCacheKeys, fmt.Sprintf("%s%d", cacheUserSubscribeUserPrefix, userId))
}
allCacheKeys := append(m.getCacheKeys(old), userSubscribeCacheKeys...)
err = m.ExecCtx(ctx, func(conn *gorm.DB) error {
db := conn
if len(tx) > 0 {
db = tx[0]
}
return db.Save(data).Error
}, allCacheKeys...)
return err
}
func (m *defaultSubscribeModel) Delete(ctx context.Context, id int64, tx ...*gorm.DB) error {
data, err := m.FindOne(ctx, id)
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
return err
}
var userIds []int64
err = m.QueryNoCacheCtx(ctx, &userIds, func(conn *gorm.DB, v interface{}) error {
return conn.Table("user_subscribe").
Where("subscribe_id = ? AND status IN (0, 1)", id).
Distinct("user_id").
Pluck("user_id", &userIds).Error
})
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
userSubscribeCacheKeys := make([]string, 0, len(userIds))
for _, userId := range userIds {
userSubscribeCacheKeys = append(userSubscribeCacheKeys, fmt.Sprintf("%s%d", cacheUserSubscribeUserPrefix, userId))
}
allCacheKeys := append(m.getCacheKeys(data), userSubscribeCacheKeys...)
err = m.ExecCtx(ctx, func(conn *gorm.DB) error {
db := conn
if len(tx) > 0 {
db = tx[0]
}
return db.Delete(&Subscribe{}, id).Error
}, allCacheKeys...)
return err
}
func (m *defaultSubscribeModel) Transaction(ctx context.Context, fn func(db *gorm.DB) error) error {
return m.TransactCtx(ctx, fn)
}