实现 gost 网关

web/services/channel_baiyin.go

@@ -9,7 +9,6 @@ import (
 	"platform/pkg/env"
 	"platform/pkg/u"
 	"platform/web/core"
-	e "platform/web/events"
 	g "platform/web/globals"
 	"platform/web/globals/orm"
 	m "platform/web/models"
@@ -17,7 +16,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/hibiken/asynq"
 	"gorm.io/gen"
 	"gorm.io/gen/field"
 )
@@ -213,41 +211,19 @@ func (s *channelBaiyinProvider) RemoveChannels(batchNo string) error {
 	return g.Redsync.WithLock(lockChannelRemoveKey(batchNo), func() error {
 		start := time.Now()
 
-		// 获取连接数据
-		channels, err := q.Channel.Where(q.Channel.BatchNo.Eq(batchNo)).Find()
+		batch, err := findUsedChanBatch(batchNo)
 		if err != nil {
-			return core.NewServErr(fmt.Sprintf("获取通道数据失败，batch：%s", batchNo), err)
+			return err
 		}
-		if len(channels) == 0 {
-			slog.Warn(fmt.Sprintf("未找到通道数据，batch：%s", batchNo))
+		if batch == nil {
+			slog.Debug("通道为空，跳过清理", "batch", batchNo)
 			return nil
 		}
 
-		proxy, err := q.Proxy.Where(q.Proxy.ID.Eq(channels[0].ProxyID)).Take()
-		if err != nil {
-			return core.NewServErr(fmt.Sprintf("获取代理数据失败，batch：%s", batchNo), err)
-		}
-
-		// 检查通道是否存在
-		chans, err := g.Redis.LRange(context.Background(), usedChansKey(proxy.ID, batchNo), 0, -1).Result()
-		if err != nil {
-			return core.NewServErr("查询使用中通道失败", err)
-		}
-		if len(chans) == 0 {
-			slog.Debug("通道为空，跳过清理", "key", usedChansKey(proxy.ID, batchNo))
-			return nil // 没有使用中通道，已经被清理过了
-		}
-
-		// 准备配置数据
-		configs := make([]*g.PortConfigsReq, len(chans))
-		for i, ch := range chans {
-			ap, err := netip.ParseAddrPort(ch)
-			if err != nil {
-				return core.NewServErr(fmt.Sprintf("解析通道数据失败: %s", ch), err)
-			}
-
+		configs := make([]*g.PortConfigsReq, len(batch.Chans))
+		for i, ch := range batch.Chans {
 			configs[i] = &g.PortConfigsReq{
-				Port:           int(ap.Port()),
+				Port:           int(ch.Port()),
 				Edge:           &[]string{},
 				AutoEdgeConfig: &g.AutoEdgeConfig{Count: u.P(0)},
 				Status:         false,
@@ -256,6 +232,11 @@ func (s *channelBaiyinProvider) RemoveChannels(batchNo string) error {
 
 		// 提交配置
 		if env.RunMode == env.RunModeProd {
+			proxy, err := q.Proxy.Where(q.Proxy.ID.Eq(batch.ProxyID)).Take()
+			if err != nil {
+				return core.NewServErr("获取代理数据失败", err)
+			}
+
 			gateway, err := proxyGateway(proxy)
 			if err != nil {
 				return core.NewServErr("创建代理网关失败", err)
@@ -271,13 +252,11 @@ func (s *channelBaiyinProvider) RemoveChannels(batchNo string) error {
 			}
 		}
 
-		// 释放端口
-		err = freeChans(proxy.ID, batchNo)
-		if err != nil {
+		if err := freeChans(batch.ProxyID, batchNo); err != nil {
 			return err
 		}
 
-		slog.Debug("清除代理端口配置", "proxy", proxy.ID, "batch", batchNo, "duration", time.Since(start).String())
+		slog.Debug("清除代理端口配置", "proxy", batch.ProxyID, "batch", batchNo, "duration", time.Since(start).String())
 		return nil
 	})
 }
@@ -335,53 +314,12 @@ func (s *channelBaiyinProvider) ClearExpiredChannels(proxyId int32) (int, error)
 	return len(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 selectProxy(count int) (*m.Proxy, g.GatewayClient, error) {
-	// 获取在线节点
-	proxies, err := q.Proxy.Where(
-		q.Proxy.Type.Eq(int(m.ProxyTypeBaiYin)),
-		q.Proxy.Status.Eq(int(m.ProxyStatusOnline)),
-	).Find()
+	proxy, err := selectProxyByType(m.ProxyTypeBaiYin, count)
 	if err != nil {
-		return nil, nil, core.NewBizErr("获取可用代理失败", err)
-	}
-	if len(proxies) == 0 {
-		return nil, nil, core.NewBizErr("无可用代理")
+		return nil, nil, err
 	}
 
-	proxyIDs := make([]int32, 0, len(proxies))
-	proxyMap := make(map[int32]*m.Proxy, len(proxies))
-	for _, item := range proxies {
-		proxyIDs = append(proxyIDs, item.ID)
-		proxyMap[item.ID] = item
-	}
-
-	// 获取最空闲节点
-	maxId := int32(0)
-	maxCount := -1
-	for _, id := range proxyIDs {
-		idCount, err := g.Redis.SCard(context.Background(), freeChansKey(id)).Result()
-		if err != nil {
-			return nil, nil, fmt.Errorf("查询可用通道数量失败: %w", err)
-		}
-
-		if idCount > int64(maxCount) {
-			maxCount = int(idCount)
-			maxId = id
-		}
-	}
-	if maxCount < count {
-		return nil, nil, core.NewBizErr("无可用代理")
-	}
-
-	proxy := proxyMap[maxId]
 	gateway, err := proxyGateway(proxy)
 	if err != nil {
 		return nil, nil, core.NewServErr("创建代理网关失败", err)
@@ -390,23 +328,6 @@ func selectProxy(count int) (*m.Proxy, g.GatewayClient, error) {
 	return proxy, gateway, nil
 }
 
-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
-}
-
 // ensureEdges 检查本地节点是否足够，如果不足从云端连入
 // 本地节点通过 Assigned = false 排除已分配节点
 // 云端节点通过 NoRepeat = true 排除已分配节点