platform/web/services/proxy.go

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{}

func proxyStatusLockKey(id int32) string {
	return fmt.Sprintf("platform:proxy:status:%d", id)
}

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
	}
	return len(keys) > 0, nil
}

func rebuildFreeChans(proxyID int32, addr netip.Addr) error {
	if err := remChans(proxyID); err != nil {
		return err
	}

	chans := make([]netip.AddrPort, 10000)
	for i := range 10000 {
		chans[i] = netip.AddrPortFrom(addr, uint16(i+10000))
	}

	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("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
}

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