shanshanzhong147
19d28a8f89
服务器用户列表缓存跨协议污染修复（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
290 lines
8.2 KiB
Go
290 lines
8.2 KiB
Go
package node
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strings"
|
||
|
||
"github.com/perfect-panel/server/pkg/tool"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
type customServerLogicModel interface {
|
||
FilterServerList(ctx context.Context, params *FilterParams) (int64, []*Server, error)
|
||
FilterNodeList(ctx context.Context, params *FilterNodeParams) (int64, []*Node, error)
|
||
ClearNodeCache(ctx context.Context, params *FilterNodeParams) error
|
||
ClearServerAllCache(ctx context.Context) error
|
||
CountNodesByIdsAndTags(ctx context.Context, nodeIds []int64, tags []string) (int64, error)
|
||
}
|
||
|
||
const (
|
||
// ServerUserListCacheKey Server User List Cache Key
|
||
ServerUserListCacheKey = "server:user:"
|
||
|
||
// ServerConfigCacheKey Server Config Cache Key
|
||
ServerConfigCacheKey = "server:config:"
|
||
)
|
||
|
||
// AllProtocols 枚举所有客户端可能携带的 protocol。
|
||
// 用户列表缓存 key 形态为 `server:user:{server_id}:{protocol}`，按 server_id
|
||
// 失效时需要按协议精确删除。SCAN 在增量 rehash 期间可能漏 key，因此采用显式枚举。
|
||
// 包含 hysteria2（兼容字段，与 hysteria 同语义），多余的 Del 是 no-op，over-deletion 安全。
|
||
var AllProtocols = []string{
|
||
"shadowsocks",
|
||
"vmess",
|
||
"vless",
|
||
"trojan",
|
||
"anytls",
|
||
"tuic",
|
||
"hysteria",
|
||
"hysteria2",
|
||
}
|
||
|
||
// ServerUserListCacheKeysForServer 返回给定 server 的所有 protocol 维度缓存 key。
|
||
// 用于 Del 路径——节点 / 订阅 / 流量统计触发缓存失效时一次性清掉该 server 下所有协议条目。
|
||
func ServerUserListCacheKeysForServer(serverId int64) []string {
|
||
keys := make([]string, 0, len(AllProtocols))
|
||
for _, protocol := range AllProtocols {
|
||
keys = append(keys, fmt.Sprintf("%s%d:%s", ServerUserListCacheKey, serverId, protocol))
|
||
}
|
||
return keys
|
||
}
|
||
|
||
// FilterParams Filter Server Params
|
||
type FilterParams struct {
|
||
Page int
|
||
Size int
|
||
Ids []int64 // Server IDs
|
||
Search string
|
||
}
|
||
|
||
type FilterNodeParams struct {
|
||
Page int // Page Number
|
||
Size int // Page Size
|
||
NodeId []int64 // Node IDs
|
||
ServerId []int64 // Server IDs
|
||
Tag []string // Tags
|
||
NodeGroupIds []int64 // Node Group IDs
|
||
Search string // Search Address or Name
|
||
Protocol string // Protocol
|
||
Preload bool // Preload Server
|
||
Enabled *bool // Enabled
|
||
}
|
||
|
||
// FilterServerList Filter Server List
|
||
func (m *customServerModel) FilterServerList(ctx context.Context, params *FilterParams) (int64, []*Server, error) {
|
||
var servers []*Server
|
||
var total int64
|
||
query := m.WithContext(ctx).Model(&Server{})
|
||
if params == nil {
|
||
params = &FilterParams{
|
||
Page: 1,
|
||
Size: 10,
|
||
}
|
||
}
|
||
if params.Search != "" {
|
||
s := "%" + params.Search + "%"
|
||
query = query.Where("`name` LIKE ? OR `address` LIKE ?", s, s)
|
||
}
|
||
if len(params.Ids) > 0 {
|
||
query = query.Where("id IN ?", params.Ids)
|
||
}
|
||
err := query.Count(&total).Order("sort ASC").Limit(params.Size).Offset((params.Page - 1) * params.Size).Find(&servers).Error
|
||
return total, servers, err
|
||
}
|
||
|
||
func (m *customServerModel) QueryServerList(ctx context.Context, ids []int64) (servers []*Server, err error) {
|
||
query := m.WithContext(ctx).Model(&Server{})
|
||
err = query.Where("id IN (?)", ids).Find(&servers).Error
|
||
return
|
||
}
|
||
|
||
// FilterNodeList Filter Node List
|
||
func (m *customServerModel) FilterNodeList(ctx context.Context, params *FilterNodeParams) (int64, []*Node, error) {
|
||
var nodes []*Node
|
||
var total int64
|
||
query := m.WithContext(ctx).Model(&Node{})
|
||
if params == nil {
|
||
params = &FilterNodeParams{
|
||
Page: 1,
|
||
Size: 10,
|
||
}
|
||
}
|
||
if params.Search != "" {
|
||
s := "%" + params.Search + "%"
|
||
query = query.Where("`name` LIKE ? OR `address` LIKE ? OR `tags` LIKE ? OR `port` LIKE ? ", s, s, s, s)
|
||
}
|
||
if len(params.NodeId) > 0 {
|
||
query = query.Where("id IN ?", params.NodeId)
|
||
}
|
||
if len(params.ServerId) > 0 {
|
||
query = query.Where("server_id IN ?", params.ServerId)
|
||
}
|
||
if len(params.Tag) > 0 {
|
||
query = query.Scopes(InSet("tags", params.Tag))
|
||
}
|
||
if len(params.NodeGroupIds) > 0 {
|
||
// Filter by node_group_ids using JSON_CONTAINS for each group ID
|
||
// Multiple group IDs: node must belong to at least one of the groups
|
||
var conditions []string
|
||
for _, gid := range params.NodeGroupIds {
|
||
conditions = append(conditions, fmt.Sprintf("JSON_CONTAINS(node_group_ids, '%d')", gid))
|
||
}
|
||
if len(conditions) > 0 {
|
||
query = query.Where("(" + strings.Join(conditions, " OR ") + ")")
|
||
}
|
||
}
|
||
// If no NodeGroupIds specified, return all nodes (including public nodes)
|
||
if params.Protocol != "" {
|
||
query = query.Where("protocol = ?", params.Protocol)
|
||
}
|
||
|
||
if params.Enabled != nil {
|
||
query = query.Where("enabled = ?", *params.Enabled)
|
||
}
|
||
|
||
if params.Preload {
|
||
query = query.Preload("Server")
|
||
}
|
||
|
||
err := query.Count(&total).Order("sort ASC").Limit(params.Size).Offset((params.Page - 1) * params.Size).Find(&nodes).Error
|
||
return total, nodes, err
|
||
}
|
||
|
||
// ClearNodeCache Clear Node Cache
|
||
func (m *customServerModel) ClearNodeCache(ctx context.Context, params *FilterNodeParams) error {
|
||
_, nodes, err := m.FilterNodeList(ctx, params)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
var cacheKeys []string
|
||
for _, node := range nodes {
|
||
cacheKeys = append(cacheKeys, ServerUserListCacheKeysForServer(node.ServerId)...)
|
||
if node.Protocol != "" {
|
||
var cursor uint64
|
||
for {
|
||
keys, newCursor, err := m.Cache.Scan(ctx, cursor, fmt.Sprintf("%s%d*", ServerConfigCacheKey, node.ServerId), 100).Result()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(keys) > 0 {
|
||
cacheKeys = append(cacheKeys, keys...)
|
||
}
|
||
cursor = newCursor
|
||
if cursor == 0 {
|
||
break
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
if len(cacheKeys) > 0 {
|
||
cacheKeys = tool.RemoveDuplicateElements(cacheKeys...)
|
||
return m.Cache.Del(ctx, cacheKeys...).Err()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ClearServerCache Clear Server Cache
|
||
func (m *customServerModel) ClearServerCache(ctx context.Context, serverId int64) error {
|
||
cacheKeys := ServerUserListCacheKeysForServer(serverId)
|
||
var cursor uint64
|
||
for {
|
||
keys, newCursor, err := m.Cache.Scan(ctx, cursor, fmt.Sprintf("%s%d*", ServerConfigCacheKey, serverId), 100).Result()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(keys) > 0 {
|
||
cacheKeys = append(cacheKeys, keys...)
|
||
}
|
||
cursor = newCursor
|
||
if cursor == 0 {
|
||
break
|
||
}
|
||
}
|
||
|
||
if len(cacheKeys) > 0 {
|
||
cacheKeys = tool.RemoveDuplicateElements(cacheKeys...)
|
||
return m.Cache.Del(ctx, cacheKeys...).Err()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (m *customServerModel) ClearServerAllCache(ctx context.Context) error {
|
||
var cursor uint64
|
||
var keys []string
|
||
prefixes := []string{ServerConfigCacheKey + "*", ServerUserListCacheKey + "*"}
|
||
for _, prefix := range prefixes {
|
||
cursor = 0
|
||
for {
|
||
scanKeys, newCursor, err := m.Cache.Scan(ctx, cursor, prefix, 999).Result()
|
||
if err != nil {
|
||
m.Logger.Error(ctx, fmt.Sprintf("ClearServerAllCache err:%v", err))
|
||
break
|
||
}
|
||
m.Logger.Info(ctx, fmt.Sprintf("ClearServerAllCache query keys:%v", scanKeys))
|
||
keys = append(keys, scanKeys...)
|
||
cursor = newCursor
|
||
if cursor == 0 {
|
||
break
|
||
}
|
||
}
|
||
}
|
||
if len(keys) > 0 {
|
||
m.Logger.Info(ctx, fmt.Sprintf("ClearServerAllCache keys:%v", keys))
|
||
return m.Cache.Del(ctx, keys...).Err()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// CountNodesByIdsAndTags 根据节点ID和标签计算启用的节点数量
|
||
func (m *customServerModel) CountNodesByIdsAndTags(ctx context.Context, nodeIds []int64, tags []string) (int64, error) {
|
||
tags = normalizeNodeTags(tags)
|
||
if len(nodeIds) == 0 && len(tags) == 0 {
|
||
return 0, nil
|
||
}
|
||
|
||
var count int64
|
||
query := m.WithContext(ctx).Model(&Node{}).Where("enabled = ?", true)
|
||
if len(nodeIds) > 0 {
|
||
query = query.Where("id IN ?", nodeIds)
|
||
}
|
||
if len(tags) > 0 {
|
||
query = query.Scopes(InSet("tags", tags))
|
||
}
|
||
|
||
err := query.Count(&count).Error
|
||
return count, err
|
||
}
|
||
|
||
func normalizeNodeTags(tags []string) []string {
|
||
cleaned := make([]string, 0, len(tags))
|
||
for _, tag := range tags {
|
||
trimmed := strings.TrimSpace(tag)
|
||
if trimmed == "" {
|
||
continue
|
||
}
|
||
cleaned = append(cleaned, trimmed)
|
||
}
|
||
return tool.RemoveDuplicateElements(cleaned...)
|
||
}
|
||
|
||
// InSet 支持多值 OR 查询
|
||
func InSet(field string, values []string) func(db *gorm.DB) *gorm.DB {
|
||
return func(db *gorm.DB) *gorm.DB {
|
||
if len(values) == 0 {
|
||
return db
|
||
}
|
||
|
||
conds := make([]string, len(values))
|
||
args := make([]interface{}, len(values))
|
||
for i, v := range values {
|
||
conds[i] = "FIND_IN_SET(?, " + field + ")"
|
||
args[i] = v
|
||
}
|
||
|
||
// 用括号包裹 OR 条件，保证外层 AND 不受影响
|
||
return db.Where("("+strings.Join(conds, " OR ")+")", args...)
|
||
}
|
||
}
|