实现代理网关管理接口

web/services/channel_baiyin.go

@@ -63,10 +63,27 @@ func (s *channelBaiyinProvider) CreateChannels(source netip.Addr, resourceId int
 	}
 	proxy := proxyResult.Proxy
 
-	// 获取可用通道
-	chans, err := lockChans(proxy.ID, batch, count)
+	// 锁内确认状态并锁定端口，避免与状态切换并发穿透
+	var chans []netip.AddrPort
+	err = g.Redsync.WithLock(proxyStatusLockKey(proxy.ID), func() error {
+		lockedProxy, err := q.Proxy.Where(q.Proxy.ID.Eq(proxy.ID)).Take()
+		if err != nil {
+			return err
+		}
+		if lockedProxy.Status != m.ProxyStatusOnline {
+			return core.NewBizErr("无可用主机，请稍后再试")
+		}
+
+		chans, err = lockChans(proxy.ID, batch, count)
+		if err != nil {
+			return core.NewBizErr("无可用通道，请稍后再试", err)
+		}
+
+		proxy = *lockedProxy
+		return nil
+	})
 	if err != nil {
-		return nil, core.NewBizErr("无可用通道，请稍后再试", err)
+		return nil, err
 	}
 
 	// 获取可用节点

web/services/proxy.go

@@ -1,65 +1,216 @@
 package services
 
 import (
+	"context"
 	"fmt"
 	"net/netip"
 	"platform/pkg/u"
 	"platform/web/core"
+	g "platform/web/globals"
 	"platform/web/globals/orm"
 	m "platform/web/models"
 	q "platform/web/queries"
+	"strconv"
 	"time"
+
+	"gorm.io/gen/field"
 )
 
 var Proxy = &proxyService{}
 
 type proxyService struct{}
 
-// AllProxies 获取所有代理
-func (s *proxyService) AllProxies(proxyType m.ProxyType, channels bool) ([]*m.Proxy, error) {
-	proxies, err := q.Proxy.Where(
-		q.Proxy.Type.Eq(int(proxyType)),
-		q.Proxy.Status.Eq(int(m.ProxyStatusOnline)),
-	).Preload(
-		q.Proxy.Channels.On(q.Channel.ExpiredAt.Gte(time.Now())),
-	).Find()
-	if err != nil {
-		return nil, err
-	}
-
-	return proxies, nil
+func proxyStatusLockKey(id int32) string {
+	return fmt.Sprintf("platform:proxy:status:%d", id)
 }
 
-// RegisterBaiyin 注册新代理服务
-func (s *proxyService) RegisterBaiyin(Name string, IP netip.Addr, username, password string) error {
-
-	// 保存代理信息
-	proxy := &m.Proxy{
-		Version: 0,
-		Mac:     Name,
-		IP:      orm.Inet{Addr: IP},
-		Secret:  u.P(fmt.Sprintf("%s:%s", username, password)),
-		Type:    m.ProxyTypeBaiYin,
-		Status:  m.ProxyStatusOnline,
+func hasUsedChans(proxyID int32) (bool, error) {
+	ctx := context.Background()
+	pattern := usedChansKey + ":" + strconv.Itoa(int(proxyID)) + ":*"
+	keys, _, err := g.Redis.Scan(ctx, 0, pattern, 1).Result()
+	if err != nil {
+		return false, err
 	}
-	if err := q.Proxy.Create(proxy); err != nil {
-		return core.NewServErr("保存通道数据失败")
+	return len(keys) > 0, nil
+}
+
+func rebuildFreeChans(proxyID int32, addr netip.Addr) error {
+	if err := remChans(proxyID); err != nil {
+		return err
 	}
 
-	// 添加可用通道到 redis
 	chans := make([]netip.AddrPort, 10000)
 	for i := range 10000 {
-		chans[i] = netip.AddrPortFrom(IP, uint16(i+10000))
+		chans[i] = netip.AddrPortFrom(addr, uint16(i+10000))
 	}
-	err := regChans(proxy.ID, chans)
+
+	if err := regChans(proxyID, chans); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *proxyService) Page(req core.PageReq) (result []*m.Proxy, count int64, err error) {
+	return q.Proxy.
+		Omit(q.Proxy.Version, q.Proxy.Meta).
+		Order(q.Proxy.CreatedAt.Desc()).
+		FindByPage(req.GetOffset(), req.GetLimit())
+}
+
+func (s *proxyService) All() (result []*m.Proxy, err error) {
+	return q.Proxy.
+		Omit(q.Proxy.Version, q.Proxy.Meta).
+		Order(q.Proxy.CreatedAt.Desc()).
+		Find()
+}
+
+type CreateProxy struct {
+	Mac    string         `json:"mac" validate:"required"`
+	IP     string         `json:"ip" validate:"required"`
+	Host   *string        `json:"host"`
+	Secret *string        `json:"secret"`
+	Type   *m.ProxyType   `json:"type"`
+	Status *m.ProxyStatus `json:"status"`
+}
+
+func (s *proxyService) Create(create *CreateProxy) error {
+	addr, err := netip.ParseAddr(create.IP)
 	if err != nil {
-		return core.NewServErr("添加通道失败", err)
+		return core.NewServErr("IP地址格式错误", err)
 	}
 
+	return q.Q.Transaction(func(tx *q.Query) error {
+		proxy := &m.Proxy{
+			Mac:    create.Mac,
+			IP:     orm.Inet{Addr: addr},
+			Host:   create.Host,
+			Secret: create.Secret,
+			Type:   u.Else(create.Type, m.ProxyTypeSelfHosted),
+			Status: u.Else(create.Status, m.ProxyStatusOffline),
+		}
+		if err := tx.Proxy.Create(proxy); err != nil {
+			return core.NewServErr("保存代理数据失败", err)
+		}
+		if err := rebuildFreeChans(proxy.ID, addr); err != nil {
+			return core.NewServErr("初始化代理通道失败", err)
+		}
+		return nil
+	})
+}
+
+type UpdateProxy struct {
+	ID     int32   `json:"id" validate:"required"`
+	Mac    *string `json:"mac"`
+	IP     *string `json:"ip"`
+	Host   *string `json:"host"`
+	Secret *string `json:"secret"`
+}
+
+func (s *proxyService) Update(update *UpdateProxy) error {
+	simples := make([]field.AssignExpr, 0)
+	hasSideEffect := false
+
+	if update.Mac != nil {
+		hasSideEffect = true
+		simples = append(simples, q.Proxy.Mac.Value(*update.Mac))
+	}
+	if update.IP != nil {
+		addr, err := netip.ParseAddr(*update.IP)
+		if err != nil {
+			return core.NewServErr("IP地址格式错误", err)
+		}
+		hasSideEffect = true
+		simples = append(simples, q.Proxy.IP.Value(orm.Inet{Addr: addr}))
+	}
+	if update.Host != nil {
+		simples = append(simples, q.Proxy.Host.Value(*update.Host))
+	}
+	if update.Secret != nil {
+		hasSideEffect = true
+		simples = append(simples, q.Proxy.Secret.Value(*update.Secret))
+	}
+
+	if len(simples) == 0 {
+		return nil
+	}
+
+	if hasSideEffect {
+		used, err := hasUsedChans(update.ID)
+		if err != nil {
+			return core.NewServErr("检查代理通道状态失败", err)
+		}
+		if used {
+			return core.NewBizErr("代理存在未关闭通道，禁止修改")
+		}
+	}
+
+	rs, err := q.Proxy.
+		Where(
+			q.Proxy.ID.Eq(update.ID),
+			q.Proxy.Status.Eq(int(m.ProxyStatusOffline)),
+		).
+		UpdateSimple(simples...)
+	if err != nil {
+		return err
+	}
+	if rs.RowsAffected == 0 {
+		return core.NewBizErr("代理未下线，禁止修改")
+	}
 	return nil
 }
 
-// UnregisterBaiyin 注销代理服务
-func (s *proxyService) UnregisterBaiyin(id int) error {
+func (s *proxyService) Remove(id int32) error {
+	used, err := hasUsedChans(id)
+	if err != nil {
+		return core.NewServErr("检查代理通道状态失败", err)
+	}
+	if used {
+		return core.NewBizErr("代理存在未关闭通道，禁止删除")
+	}
+
+	rs, err := q.Proxy.
+		Where(
+			q.Proxy.ID.Eq(id),
+			q.Proxy.Status.Eq(int(m.ProxyStatusOffline)),
+		).
+		UpdateColumn(q.Proxy.DeletedAt, time.Now())
+	if err != nil {
+		return err
+	}
+	if rs.RowsAffected == 0 {
+		return core.NewBizErr("代理未下线，禁止删除")
+	}
+	if err := remChans(id); err != nil {
+		return core.NewServErr("注销代理通道失败", err)
+	}
 	return nil
 }
+
+type UpdateProxyStatus struct {
+	ID     int32         `json:"id" validate:"required"`
+	Status m.ProxyStatus `json:"status" validate:"required"`
+}
+
+func (s *proxyService) UpdateStatus(update *UpdateProxyStatus) error {
+	return g.Redsync.WithLock(proxyStatusLockKey(update.ID), func() error {
+		proxy, err := q.Proxy.Where(q.Proxy.ID.Eq(update.ID)).Take()
+		if err != nil {
+			return err
+		}
+
+		if proxy.Status == update.Status {
+			return nil
+		}
+
+		if update.Status == m.ProxyStatusOnline {
+			if err := rebuildFreeChans(proxy.ID, proxy.IP.Addr); err != nil {
+				return core.NewServErr("初始化代理通道失败", err)
+			}
+		}
+
+		_, err = q.Proxy.
+			Where(q.Proxy.ID.Eq(update.ID)).
+			UpdateSimple(q.Proxy.Status.Value(int(update.Status)))
+		return err
+	})
+}