platform/web/services/channel.go

package services

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"math/rand/v2"
	"net/netip"
	"platform/pkg/u"
	"platform/web/core"
	e "platform/web/events"
	g "platform/web/globals"
	"platform/web/globals/orm"
	m "platform/web/models"
	q "platform/web/queries"
	"strconv"
	"strings"
	"time"

	"github.com/hibiken/asynq"
	"github.com/redis/go-redis/v9"
	"gorm.io/gen"
	"gorm.io/gen/field"
)

// 通道服务
var Channel = &channelServer{
	provider: &channelGostProvider{},
}

type channelProvider interface {
	selectProxy(count int) (*m.Proxy, error)
	prepareCreate(ctx *channelCreateContext) (*channelCreateResult, error)
	removeRemote(batchNo string, batch *usedChanBatch) error
}

type channelServer struct {
	provider channelProvider
}

func (s *channelServer) CreateChannels(source netip.Addr, resourceNo string, authWhitelist bool, authPassword bool, count int, edgeFilter *EdgeFilter) ([]*m.Channel, error) {
	var area *m.Area
	if edgeFilter.AreaID != nil {
		var err error
		area, err = Area.Get(*edgeFilter.AreaID)
		if err != nil {
			return nil, err
		}
		if err := validateChannelArea(area); err != nil {
			return nil, err
		}
	}

	now := time.Now()
	batchNo := ID.GenReadable("bat")
	var channels []*m.Channel

	var whitelistText *string
	err := g.Redsync.WithLock(lockChannelCreateKey(resourceNo), func() error {
		resource, whitelists, err := ensure(now, source, resourceNo, authWhitelist, count)
		if err != nil {
			return err
		}

		if authWhitelist {
			joined := strings.Join(whitelists, ",")
			whitelistText = &joined
		}

		expire := now.Add(resource.Live)
		proxy, err := s.provider.selectProxy(count)
		if err != nil {
			return err
		}

		ports, err := selectPorts(proxy.ID, batchNo, count, expire)
		if err != nil {
			return err
		}

		createCtx := &channelCreateContext{
			Now:           now,
			Source:        source,
			Resource:      resource,
			Proxy:         proxy,
			BatchNo:       batchNo,
			Ports:         ports,
			Expire:        expire,
			Count:         count,
			Filter:        edgeFilter,
			Area:          area,
			AuthWhitelist: authWhitelist,
			AuthPassword:  authPassword,
			Whitelists:    whitelists,
			WhitelistText: whitelistText,
		}

		result, err := s.provider.prepareCreate(createCtx)
		if err != nil {
			return err
		}
		if result.applyRemote != nil {
			if err := result.applyRemote(); err != nil {
				return err
			}
		}
		if err := persistChannelCreate(createCtx, result.Channels); err != nil {
			return err
		}

		channels = result.Channels
		return nil
	})
	if err != nil {
		return nil, err
	}

	return channels, nil
}

func (s *channelServer) RemoveChannels(batch string) error {
	return g.Redsync.WithLock(lockChannelRemoveKey(batch), func() error {
		start := time.Now()

		usedBatch, err := findUsedChanBatch(batch)
		if err != nil {
			return err
		}
		if usedBatch == nil {
			slog.Debug("通道为空，跳过清理", "batch", batch)
			return nil
		}

		if err := s.provider.removeRemote(batch, usedBatch); err != nil {
			return err
		}
		if err := freeChans(usedBatch.ProxyID, batch); err != nil {
			return err
		}

		slog.Debug("清除通道配置", "proxy", usedBatch.ProxyID, "batch", batch, "duration", time.Since(start).String())
		return nil
	})
}

func (s *channelServer) ClearExpiredChannels(proxyId int32) (int, error) {
	batchSet, err := findExpiredChannelBatches(proxyId, time.Now())
	if err != nil {
		return 0, err
	}

	slog.Info("批量清理过期通道", "count", len(batchSet))
	for batchNo := range batchSet {
		if err := s.RemoveChannels(batchNo); err != nil {
			slog.Error("清理过期通道失败", "batch", batchNo, "error", err)
		}
	}

	return len(batchSet), nil
}

type channelCreateContext struct {
	Now           time.Time
	Source        netip.Addr
	Resource      *ResourceView
	Proxy         *m.Proxy
	BatchNo       string
	Ports         []netip.AddrPort
	Expire        time.Time
	Count         int
	Filter        *EdgeFilter
	Area          *m.Area
	AuthWhitelist bool
	AuthPassword  bool
	Whitelists    []string
	WhitelistText *string
}

type channelCreateResult struct {
	Channels    []*m.Channel
	applyRemote func() error
}

func newBaseChannel(ctx *channelCreateContext, port uint16) *m.Channel {
	prov, city := areaProvinceCity(ctx.Area)
	return &m.Channel{
		UserID:     ctx.Resource.User.ID,
		ResourceID: ctx.Resource.ID,
		BatchNo:    ctx.BatchNo,
		ProxyID:    ctx.Proxy.ID,
		Host:       u.Else(ctx.Proxy.Host, ctx.Proxy.IP.String()),
		Port:       port,
		FilterISP:  ctx.Filter.Isp,
		FilterProv: prov,
		FilterCity: city,
		ExpiredAt:  ctx.Expire,
		Proxy:      ctx.Proxy,
	}
}

func applyChannelAuth(ctx *channelCreateContext, channel *m.Channel) (username string, password string, ok bool) {
	if ctx.AuthWhitelist {
		channel.Whitelists = ctx.WhitelistText
	}
	if !ctx.AuthPassword {
		return "", "", false
	}

	username, password = genPassPair()
	channel.Username = &username
	channel.Password = &password
	return username, password, true
}

func persistChannelCreate(ctx *channelCreateContext, channels []*m.Channel) error {
	prov, city := areaProvinceCity(ctx.Area)
	return q.Q.Transaction(func(tx *q.Query) error {
		var (
			result gen.ResultInfo
			err    error
		)
		switch ctx.Resource.Type {
		case m.ResourceTypeShort:
			result, err = tx.ResourceShort.
				Where(
					tx.ResourceShort.ID.Eq(*ctx.Resource.ShortId),
					tx.ResourceShort.Used.Eq(ctx.Resource.Used),
					tx.ResourceShort.Daily.Eq(ctx.Resource.Daily),
				).
				UpdateSimple(
					tx.ResourceShort.Used.Add(int32(ctx.Count)),
					tx.ResourceShort.Daily.Value(int32(ctx.Resource.Today+ctx.Count)),
					tx.ResourceShort.LastAt.Value(ctx.Now),
				)
		case m.ResourceTypeLong:
			result, err = tx.ResourceLong.
				Where(
					tx.ResourceLong.ID.Eq(*ctx.Resource.LongId),
					tx.ResourceLong.Used.Eq(ctx.Resource.Used),
					tx.ResourceLong.Daily.Eq(ctx.Resource.Daily),
				).
				UpdateSimple(
					tx.ResourceLong.Used.Add(int32(ctx.Count)),
					tx.ResourceLong.Daily.Value(int32(ctx.Resource.Today+ctx.Count)),
					tx.ResourceLong.LastAt.Value(ctx.Now),
				)
		default:
			return core.NewBizErr("套餐类型不正确，无法更新")
		}
		if err != nil {
			return core.NewServErr("更新套餐使用记录失败", err)
		}
		if result.RowsAffected == 0 {
			return core.NewBizErr("套餐状态已过期")
		}

		if err := tx.Channel.Omit(field.AssociationFields).Create(channels...); err != nil {
			return core.NewServErr("保存通道失败", err)
		}

		if err := tx.LogsUserUsage.Create(&m.LogsUserUsage{
			UserID:     ctx.Resource.User.ID,
			ResourceID: ctx.Resource.ID,
			BatchNo:    ctx.BatchNo,
			Count:      int32(ctx.Count),
			ISP:        u.X(ctx.Filter.Isp.String()),
			Prov:       prov,
			City:       city,
			IP:         orm.Inet{Addr: ctx.Source},
			Time:       ctx.Now,
		}); err != nil {
			return core.NewServErr("保存用户使用记录失败", err)
		}

		return nil
	})
}

func validateChannelArea(area *m.Area) error {
	if area == nil {
		return nil
	}
	switch area.Level {
	case m.AreaLevelProvince:
		return nil
	case m.AreaLevelCity:
		if area.ParentID == nil || area.Parent == nil {
			return core.NewServErr("地区数据异常", nil)
		}
		return nil
	default:
		return core.NewBizErr("地区层级不支持")
	}
}

func areaProvinceCity(area *m.Area) (prov *string, city *string) {
	if area == nil {
		return nil, nil
	}
	switch area.Level {
	case m.AreaLevelProvince:
		return u.P(area.Name), nil
	case m.AreaLevelCity:
		return u.P(area.Parent.Name), u.P(area.Name)
	default:
		return nil, nil
	}
}

func findExpiredChannelBatches(proxyId int32, now time.Time) (map[string]struct{}, error) {
	keys, err := g.Redis.Keys(context.Background(), usedChansKey(proxyId, "*")).Result()
	if err != nil {
		return nil, core.NewServErr("查询使用中通道失败", err)
	}
	if len(keys) == 0 {
		return map[string]struct{}{}, nil
	}

	batchList := make([]string, len(keys))
	batchSet := make(map[string]struct{}, len(keys))
	for i, key := range keys {
		parsed, err := parseUsedChanKey(key)
		if err != nil {
			return nil, err
		}
		batchList[i] = parsed.BatchNo
		batchSet[parsed.BatchNo] = struct{}{}
	}

	var batchQueried []struct{ BatchNo string }
	err = q.Channel.
		Select(q.Channel.BatchNo).
		Where(
			q.Channel.BatchNo.In(batchList...),
			q.Channel.ExpiredAt.Gte(now.UTC()),
		).
		Group(q.Channel.BatchNo).
		Scan(&batchQueried)
	if err != nil {
		return nil, core.NewServErr("查询过期通道失败", err)
	}
	for _, batch := range batchQueried {
		delete(batchSet, batch.BatchNo)
	}

	return batchSet, nil
}

func lockChannelCreateKey(resourceNo string) string {
	return fmt.Sprintf("platform:channel:create:%s", resourceNo)
}

func lockChannelRemoveKey(bid string) string {
	return fmt.Sprintf("platform:batch:remove_expired:%s", bid)
}

func selectPorts(proxyId int32, batchNo string, count int, expire time.Time) ([]netip.AddrPort, error) {
	chans, err := lockChans(proxyId, batchNo, count)
	if err != nil {
		return nil, core.NewBizErr("无可用通道，请稍后再试", err)
	}

	_, err = g.Asynq.Enqueue(
		e.NewRemoveChannel(batchNo),
		asynq.ProcessAt(expire),
	)
	if err != nil {
		return nil, core.NewServErr("注册异步关闭通道任务失败", err)
	}

	return chans, nil
}

func selectProxyByType(proxyType m.ProxyType, count int) (*m.Proxy, error) {
	proxies, err := q.Proxy.Where(
		q.Proxy.Type.Eq(int(proxyType)),
		q.Proxy.Status.Eq(int(m.ProxyStatusOnline)),
	).Find()
	if err != nil {
		return nil, core.NewBizErr("获取可用代理失败", err)
	}
	if len(proxies) == 0 {
		return nil, core.NewBizErr("无可用代理")
	}

	var bestProxy *m.Proxy
	maxCount := -1
	for _, proxy := range proxies {
		idCount, err := g.Redis.SCard(context.Background(), freeChansKey(proxy.ID)).Result()
		if err != nil {
			return nil, core.NewServErr("查询可用通道数量失败", err)
		}
		if idCount > int64(maxCount) {
			maxCount = int(idCount)
			bestProxy = proxy
		}
	}
	if maxCount < count {
		return nil, core.NewBizErr("无可用代理")
	}

	return bestProxy, nil
}

func (s *channelServer) RefreshEdges() error {

	// 仅白银网关支持边缘节点刷新，GOST 不参与此流程。
	proxies, err := q.Proxy.Where(
		q.Proxy.Type.Eq(int(m.ProxyTypeBaiYin)),
		q.Proxy.Status.Eq(int(m.ProxyStatusOnline)),
	).Find()
	if err != nil {
		return fmt.Errorf("查询网关失败: %w", err)
	}

	for _, proxy := range proxies {
		gateway, err := proxyGateway(proxy)
		if err != nil {
			return core.NewServErr("创建代理网关失败", err)
		}

		// 选取随机节点
		edges, err := gateway.GatewayEdge(&g.GatewayEdgeReq{
			Assigned: u.P(false),
			Limit:    u.P(1000),
		})
		if err != nil {
			return fmt.Errorf("获取边缘节点失败: %w", err)
		}

		// 提交断开配置
		edgeIds := make([]string, 0, len(edges))
		for id, _ := range edges {
			edgeIds = append(edgeIds, id)
		}
		g.Cloud.CloudDisconnect(&g.CloudDisconnectReq{
			Uuid: proxy.Mac,
			Edge: &edgeIds,
		})
	}

	return nil
}

// 授权方式
type ChannelAuthType int

const (
	ChannelAuthTypeIp ChannelAuthType = iota + 1
	ChannelAuthTypePass
)

// 生成用户名和密码对
func genPassPair() (string, string) {
	var alphabet = []rune("abcdefghjkmnpqrstuvwxyz")
	var numbers = []rune("23456789")

	var username = make([]rune, 6)
	var password = make([]rune, 6)
	for i := range 6 {
		if i < 2 {
			username[i] = alphabet[rand.N(len(alphabet))]
		} else {
			username[i] = numbers[rand.N(len(numbers))]
		}
		password[i] = numbers[rand.N(len(numbers))]
	}

	return string(username), string(password)
}

func FindResourceNoById(resourceId int32) (string, error) {
	resource, err := q.Resource.
		Select(q.Resource.ResourceNo).
		Where(q.Resource.ID.Eq(resourceId)).
		Take()
	if err != nil {
		return "", ErrResourceNotExist
	}
	return u.Z(resource.ResourceNo), nil
}

// 查找资源
func findResourceViewByNo(resourceNo string, now time.Time) (*ResourceView, error) {
	resource, err := q.Resource.
		Preload(field.Associations).
		Where(
			q.Resource.ResourceNo.Eq(resourceNo),
			q.Resource.Active.Is(true),
		).
		Take()
	if err != nil {
		return nil, ErrResourceNotExist
	}
	if resource.User == nil {
		return nil, ErrResourceNotExist
	}
	var info = &ResourceView{
		ID:      resource.ID,
		User:    *resource.User,
		Active:  resource.Active,
		Type:    resource.Type,
		CheckIP: resource.CheckIP,
	}

	switch resource.Type {
	case m.ResourceTypeShort:
		var sub = resource.Short
		info.ShortId = &sub.ID
		info.ExpireAt = sub.ExpireAt
		info.Live = time.Duration(sub.Live) * time.Minute
		info.Mode = sub.Type
		info.Quota = sub.Quota
		info.Used = sub.Used
		info.Daily = sub.Daily
		info.LastAt = sub.LastAt
		if sub.LastAt != nil && u.IsSameDate(*sub.LastAt, now) {
			info.Today = int(sub.Daily)
		}

	case m.ResourceTypeLong:
		var sub = resource.Long
		info.LongId = &sub.ID
		info.ExpireAt = sub.ExpireAt
		info.Live = time.Duration(sub.Live) * time.Minute
		info.Mode = sub.Type
		info.Quota = sub.Quota
		info.Used = sub.Used
		info.Daily = sub.Daily
		info.LastAt = sub.LastAt
		if sub.LastAt != nil && u.IsSameDate(*sub.LastAt, now) {
			info.Today = int(sub.Daily)
		}
	}

	if info.Mode == m.ResourceModeTime && info.ExpireAt == nil {
		return nil, errors.New("检查套餐获取时间失败")
	}

	return info, nil
}

// ResourceView 套餐数据的简化视图，便于直接获取主要数据
type ResourceView struct {
	ID       int32
	User     m.User
	Active   bool
	Type     m.ResourceType
	ShortId  *int32
	LongId   *int32
	Live     time.Duration
	Mode     m.ResourceMode
	Quota    int32
	ExpireAt *time.Time
	Used     int32
	Daily    int32
	LastAt   *time.Time
	Today    int // 今日用量
	CheckIP  bool
}

// 检查用户是否可提取
func ensure(now time.Time, source netip.Addr, resourceNo string, authWhitelist bool, count int) (*ResourceView, []string, error) {
	if count > 400 {
		return nil, nil, core.NewBizErr("单次最多提取 400 个")
	}

	// 获取用户套餐
	resource, err := findResourceViewByNo(resourceNo, now)
	if err != nil {
		return nil, nil, err
	}

	// 检查用户
	user := resource.User
	if user.IDToken == nil || *user.IDToken == "" {
		return nil, nil, core.NewBizErr("账号未实名")
	}

	// 获取用户白名单并检查用户 ip 地址
	whitelists, err := q.Whitelist.Where(
		q.Whitelist.UserID.Eq(user.ID),
	).Find()
	if err != nil {
		return nil, nil, err
	}

	if authWhitelist && len(whitelists) == 0 {
		return nil, nil, core.NewBizErr("当前白名单为空，请先添加白名单")
	}

	ips := make([]string, len(whitelists))
	pass := false
	for i, item := range whitelists {
		ips[i] = item.IP.String()
		if item.IP.Addr == source {
			pass = true
		}
	}
	if resource.CheckIP && !pass {
		return nil, nil, core.NewBizErr(fmt.Sprintf("IP 地址 %s 不在白名单内", source.String()))
	}

	// 检查套餐使用情况
	switch resource.Mode {
	default:
		return nil, nil, core.NewBizErr("不支持的套餐模式")

	// 包时
	case m.ResourceModeTime:
		// 检查过期时间
		if resource.ExpireAt.Before(now) {
			return nil, nil, ErrResourceExpired
		}
		// 检查每日限额
		if count+resource.Today > int(resource.Quota) {
			return nil, nil, ErrResourceDailyLimit
		}

	// 包量
	case m.ResourceModeQuota:
		// 检查可用配额
		if int(resource.Quota)-int(resource.Used) < count {
			return nil, nil, ErrResourceExhausted
		}
	}

	return resource, ips, nil
}

func freeChansKey(proxy int32) string {
	return "channel:free:" + strconv.Itoa(int(proxy))
}

func usedChansKey(proxy int32, batch string) string {
	return "channel:used:" + strconv.Itoa(int(proxy)) + ":" + batch
}

type usedChanBatch struct {
	ProxyID int32
	Chans   []netip.AddrPort
}

type usedChanKey struct {
	ProxyID int32
	BatchNo string
}

func findUsedChanBatch(batch string) (*usedChanBatch, error) {
	keys, err := g.Redis.Keys(context.Background(), "channel:used:*:"+batch).Result()
	if err != nil {
		return nil, core.NewServErr("查询使用中通道失败", err)
	}

	key, ok, err := selectUsedChanBatchKey(batch, keys)
	if err != nil {
		return nil, err
	}
	if !ok {
		return nil, nil
	}

	chans, err := g.Redis.LRange(context.Background(), key, 0, -1).Result()
	if err != nil {
		return nil, core.NewServErr("查询使用中通道失败", err)
	}

	return parseUsedChanBatch(key, chans)
}

func selectUsedChanBatchKey(batch string, keys []string) (string, bool, error) {
	switch len(keys) {
	case 0:
		return "", false, nil
	case 1:
		return keys[0], true, nil
	default:
		slog.Error("batchNo 全局唯一约束被破坏", "batch", batch, "keys", keys)
		return "", false, core.NewServErr(
			fmt.Sprintf("检测到重复 usedChans 键，batchNo 全局唯一被破坏: %s", batch),
			fmt.Errorf("keys=%s", strings.Join(keys, ",")),
		)
	}
}

func parseUsedChanBatch(key string, chans []string) (*usedChanBatch, error) {
	parsed, err := parseUsedChanKey(key)
	if err != nil {
		return nil, err
	}

	addrs := make([]netip.AddrPort, len(chans))
	for i, ch := range chans {
		addr, err := netip.ParseAddrPort(ch)
		if err != nil {
			return nil, core.NewServErr(fmt.Sprintf("解析通道数据失败: %s", ch), err)
		}
		addrs[i] = addr
	}

	return &usedChanBatch{
		ProxyID: parsed.ProxyID,
		Chans:   addrs,
	}, nil
}

func parseUsedChanKey(key string) (*usedChanKey, error) {
	parts := strings.Split(key, ":")
	if len(parts) != 4 {
		return nil, core.NewServErr(fmt.Sprintf("使用中通道键格式错误: %s", key), nil)
	}

	proxyID, err := strconv.Atoi(parts[2])
	if err != nil {
		return nil, core.NewServErr(fmt.Sprintf("使用中通道键格式错误: %s", key), err)
	}

	return &usedChanKey{
		ProxyID: int32(proxyID),
		BatchNo: parts[3],
	}, nil
}

// 扩容通道
func regChans(proxy int32, chans []netip.AddrPort) error {
	strs := make([]any, len(chans))
	for i, ch := range chans {
		strs[i] = ch.String()
	}

	key := freeChansKey(proxy)
	err := g.Redis.SAdd(context.Background(), key, strs...).Err()
	if err != nil {
		return fmt.Errorf("扩容通道失败: %w", err)
	}
	return nil
}

// 缩容通道
func remChans(proxy int32) error {
	key := freeChansKey(proxy)
	err := g.Redis.Del(context.Background(), key).Err()
	if err != nil {
		return fmt.Errorf("缩容通道失败: %w", err)
	}
	return nil
}

// 取用通道
func lockChans(proxy int32, batch string, count int) ([]netip.AddrPort, error) {
	chans, err := RedisScriptLockChans.Run(
		context.Background(),
		g.Redis,
		[]string{
			freeChansKey(proxy),
			usedChansKey(proxy, batch),
		},
		count,
	).StringSlice()
	if err != nil {
		return nil, fmt.Errorf("获取通道失败: %w", err)
	}

	addrs := make([]netip.AddrPort, len(chans))
	for i, ch := range chans {
		addr, err := netip.ParseAddrPort(ch)
		if err != nil {
			return nil, fmt.Errorf("解析通道数据失败: %w", err)
		}
		addrs[i] = addr
	}

	return addrs, nil
}

var RedisScriptLockChans = redis.NewScript(`
local free_key  = KEYS[1]
local batch_key = KEYS[2]
local count     = tonumber(ARGV[1])

local free_count = redis.call("SCARD", free_key)
if count <= 0 or free_count < count then
    return nil
end

local ports = redis.call("SPOP", free_key, count)
redis.call("RPUSH", batch_key, unpack(ports))

return ports
`)

// 归还通道
func freeChans(proxy int32, batch string) error {
	err := RedisScriptFreeChans.Run(
		context.Background(),
		g.Redis,
		[]string{
			freeChansKey(proxy),
			usedChansKey(proxy, batch),
		},
	).Err()
	if err != nil {
		return core.NewBizErr("释放通道失败", err)
	}

	return nil
}

var RedisScriptFreeChans = redis.NewScript(`
local free_key  = KEYS[1]
local batch_key = KEYS[2]

local chans = redis.call("LRANGE", batch_key, 0, -1)
if #chans == 0 then
    return 1
end

redis.call("SADD", free_key, unpack(chans))
redis.call("DEL", batch_key)

return 1
`)

// 节点筛选条件
type EdgeFilter struct {
	Isp    *m.EdgeISP `json:"isp"`
	AreaID *int32     `json:"area_id"`
}

func (f *EdgeFilter) IsEmpty() bool {
	if f == nil {
		return true
	}

	return u.X(f.Isp.String()) == nil && f.AreaID == nil
}

// 错误信息
var (
	ErrResourceNotExist   = core.NewBizErr("套餐不存在")
	ErrResourceExhausted  = core.NewBizErr("套餐已用完")
	ErrResourceExpired    = core.NewBizErr("套餐已过期")
	ErrResourceDailyLimit = core.NewBizErr("套餐每日配额已用完")
)