hi-server/internal/model/node/model.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",
	"simnet",
}

// 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...)
	}
}