Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a964fe4d69 | |||
| 6db3caaecb | |||
| fd475d3e63 |
@@ -19,7 +19,7 @@ WORKDIR /app
|
||||
|
||||
ENV TZ=Asia/Shanghai
|
||||
|
||||
RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.tuna.tsinghua.edu.cn/g' /etc/apk/repositories
|
||||
RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories
|
||||
RUN apk add --no-cache ca-certificates tzdata
|
||||
|
||||
COPY --from=builder /build/bin/platform_linux_amd64 /app/platform
|
||||
|
||||
17
README.md
17
README.md
@@ -1,8 +1,10 @@
|
||||
## TODO
|
||||
|
||||
交易信息持久化
|
||||
proxy 的删除和更新,锁粒度应该有问题
|
||||
|
||||
用户请求需要检查数据权限
|
||||
最低价格 0.01
|
||||
|
||||
交易信息持久化
|
||||
|
||||
用反射实现环境变量解析,以简化函数签名
|
||||
|
||||
@@ -10,12 +12,8 @@
|
||||
|
||||
分离 task 的客户端,支持多进程(prefork 必要!)
|
||||
|
||||
jsonb 类型转换问题,考虑一个高效的 any 到 struct 转换工具
|
||||
|
||||
慢速请求底层调用埋点监控
|
||||
|
||||
数据库转模型文件
|
||||
|
||||
冷数据迁移方案
|
||||
|
||||
## 开发环境
|
||||
@@ -49,13 +47,6 @@ jsonb 类型转换问题,考虑一个高效的 any 到 struct 转换工具
|
||||
3. 异步回调事件,收到支付成功事件后自动完成订单
|
||||
4. 用户退出支付界面,客户端主动发起关闭订单
|
||||
|
||||
### 产品字典表
|
||||
|
||||
| 代码 | 产品 |
|
||||
| ----- | ------------ |
|
||||
| short | 短效动态代理 |
|
||||
| long | 长效动态代理 |
|
||||
|
||||
### 节点分配与存储逻辑
|
||||
|
||||
提取:
|
||||
|
||||
@@ -31,6 +31,15 @@ services:
|
||||
ports:
|
||||
- "5433:5432"
|
||||
|
||||
asynqmon:
|
||||
image: hibiken/asynqmon:latest
|
||||
environment:
|
||||
- REDIS_ADDR=redis:6379
|
||||
ports:
|
||||
- "9800:8080"
|
||||
depends_on:
|
||||
- redis
|
||||
|
||||
volumes:
|
||||
postgres_data:
|
||||
redis_data:
|
||||
|
||||
@@ -126,7 +126,8 @@ insert into permission (name, description, sort) values
|
||||
('channel', 'IP', 11),
|
||||
('trade', '交易', 12),
|
||||
('bill', '账单', 13),
|
||||
('balance_activity', '余额变动', 14);
|
||||
('balance_activity', '余额变动', 14),
|
||||
('proxy', '代理', 15);
|
||||
|
||||
-- --------------------------
|
||||
-- level 2
|
||||
@@ -189,6 +190,11 @@ insert into permission (parent_id, name, description, sort) values
|
||||
((select id from permission where name = 'channel' and deleted_at is null), 'channel:read', '读取 IP 列表', 1),
|
||||
((select id from permission where name = 'channel' and deleted_at is null), 'channel:write', '写入 IP', 2);
|
||||
|
||||
-- proxy 子权限
|
||||
insert into permission (parent_id, name, description, sort) values
|
||||
((select id from permission where name = 'proxy' and deleted_at is null), 'proxy:read', '读取代理列表', 1),
|
||||
((select id from permission where name = 'proxy' and deleted_at is null), 'proxy:write', '写入代理', 2);
|
||||
|
||||
-- trade 子权限
|
||||
insert into permission (parent_id, name, description, sort) values
|
||||
((select id from permission where name = 'trade' and deleted_at is null), 'trade:read', '读取交易列表', 1),
|
||||
@@ -211,6 +217,10 @@ insert into permission (parent_id, name, description, sort) values
|
||||
insert into permission (parent_id, name, description, sort) values
|
||||
((select id from permission where name = 'product_sku:write' and deleted_at is null), 'product_sku:write:status', '更改产品套餐状态', 1);
|
||||
|
||||
-- proxy:write 子权限
|
||||
insert into permission (parent_id, name, description, sort) values
|
||||
((select id from permission where name = 'proxy:write' and deleted_at is null), 'proxy:write:status', '更改代理状态', 1);
|
||||
|
||||
-- resource:short 子权限
|
||||
insert into permission (parent_id, name, description, sort) values
|
||||
((select id from permission where name = 'resource:short' and deleted_at is null), 'resource:short:read', '读取用户短效动态套餐列表', 1);
|
||||
|
||||
@@ -62,6 +62,11 @@ const (
|
||||
ScopeChannelReadOfUser = string("channel:read:of_user") // 读取指定用户的 IP 列表
|
||||
ScopeChannelWrite = string("channel:write") // 写入 IP
|
||||
|
||||
ScopeProxy = string("proxy") // 代理
|
||||
ScopeProxyRead = string("proxy:read") // 读取代理列表
|
||||
ScopeProxyWrite = string("proxy:write") // 写入代理
|
||||
ScopeProxyWriteStatus = string("proxy:write:status") // 更改代理状态
|
||||
|
||||
ScopeTrade = string("trade") // 交易
|
||||
ScopeTradeRead = string("trade:read") // 读取交易列表
|
||||
ScopeTradeReadOfUser = string("trade:read:of_user") // 读取指定用户的交易列表
|
||||
|
||||
@@ -15,12 +15,12 @@ func PageAdminByAdmin(c *fiber.Ctx) error {
|
||||
return err
|
||||
}
|
||||
|
||||
var req PageAdminsReq
|
||||
var req core.PageReq
|
||||
if err := g.Validator.ParseBody(c, &req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
list, total, err := s.Admin.Page(req.PageReq)
|
||||
list, total, err := s.Admin.Page(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -33,10 +33,6 @@ func PageAdminByAdmin(c *fiber.Ctx) error {
|
||||
})
|
||||
}
|
||||
|
||||
type PageAdminsReq struct {
|
||||
core.PageReq
|
||||
}
|
||||
|
||||
func AllAdminByAdmin(c *fiber.Ctx) error {
|
||||
_, err := auth.GetAuthCtx(c).PermitAdmin(core.ScopeAdminRead)
|
||||
if err != nil {
|
||||
|
||||
@@ -201,7 +201,7 @@ func CreateChannel(c *fiber.Ctx) error {
|
||||
req.AuthType == s.ChannelAuthTypeIp,
|
||||
req.AuthType == s.ChannelAuthTypePass,
|
||||
req.Count,
|
||||
s.EdgeFilter{
|
||||
&s.EdgeFilter{
|
||||
Isp: isp,
|
||||
Prov: req.Prov,
|
||||
City: req.City,
|
||||
|
||||
@@ -1,61 +1,123 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"net/netip"
|
||||
"platform/pkg/env"
|
||||
"platform/web/auth"
|
||||
"platform/web/core"
|
||||
"platform/web/globals"
|
||||
g "platform/web/globals"
|
||||
s "platform/web/services"
|
||||
"time"
|
||||
|
||||
"github.com/gofiber/fiber/v2"
|
||||
)
|
||||
|
||||
func DebugRegisterProxyBaiYin(c *fiber.Ctx) error {
|
||||
if env.RunMode != env.RunModeDev {
|
||||
return fiber.ErrNotFound
|
||||
}
|
||||
|
||||
err := s.Proxy.RegisterBaiyin("1a:2b:3c:4d:5e:6f", netip.AddrFrom4([4]byte{127, 0, 0, 1}), "test", "test")
|
||||
if err != nil {
|
||||
return core.NewServErr("注册失败", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// 注册白银代理网关
|
||||
func ProxyRegisterBaiYin(c *fiber.Ctx) error {
|
||||
_, err := auth.GetAuthCtx(c).PermitOfficialClient()
|
||||
func PageProxyByAdmin(c *fiber.Ctx) error {
|
||||
_, err := auth.GetAuthCtx(c).PermitAdmin(core.ScopeProxyRead)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
req := new(RegisterProxyBaiyinReq)
|
||||
err = globals.Validator.ParseBody(c, req)
|
||||
var req core.PageReq
|
||||
if err := g.Validator.ParseBody(c, &req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
list, total, err := s.Proxy.Page(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
addr, err := netip.ParseAddr(req.IP)
|
||||
if err != nil {
|
||||
return core.NewServErr("IP地址格式错误", err)
|
||||
}
|
||||
|
||||
err = s.Proxy.RegisterBaiyin(req.Name, addr, req.Username, req.Password)
|
||||
if err != nil {
|
||||
return core.NewServErr("注册失败", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
return c.JSON(core.PageResp{
|
||||
List: list,
|
||||
Total: int(total),
|
||||
Page: req.GetPage(),
|
||||
Size: req.GetSize(),
|
||||
})
|
||||
}
|
||||
|
||||
type RegisterProxyBaiyinReq struct {
|
||||
Name string `json:"name" validate:"required"`
|
||||
IP string `json:"ip" validate:"required"`
|
||||
Username string `json:"username" validate:"required"`
|
||||
Password string `json:"password" validate:"required"`
|
||||
func AllProxyByAdmin(c *fiber.Ctx) error {
|
||||
_, err := auth.GetAuthCtx(c).PermitAdmin(core.ScopeProxyRead)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
list, err := s.Proxy.All()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.JSON(list)
|
||||
}
|
||||
|
||||
func CreateProxy(c *fiber.Ctx) error {
|
||||
_, err := auth.GetAuthCtx(c).PermitAdmin(core.ScopeProxyWrite)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var req s.CreateProxy
|
||||
if err := g.Validator.ParseBody(c, &req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := s.Proxy.Create(&req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.JSON(nil)
|
||||
}
|
||||
|
||||
func UpdateProxy(c *fiber.Ctx) error {
|
||||
_, err := auth.GetAuthCtx(c).PermitAdmin(core.ScopeProxyWrite)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var req s.UpdateProxy
|
||||
if err := g.Validator.ParseBody(c, &req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := s.Proxy.Update(&req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.JSON(nil)
|
||||
}
|
||||
|
||||
func UpdateProxyStatus(c *fiber.Ctx) error {
|
||||
_, err := auth.GetAuthCtx(c).PermitAdmin(core.ScopeProxyWriteStatus)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var req s.UpdateProxyStatus
|
||||
if err := g.Validator.ParseBody(c, &req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := s.Proxy.UpdateStatus(&req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.JSON(nil)
|
||||
}
|
||||
|
||||
func RemoveProxy(c *fiber.Ctx) error {
|
||||
_, err := auth.GetAuthCtx(c).PermitAdmin(core.ScopeProxyWrite)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var req core.IdReq
|
||||
if err := g.Validator.ParseBody(c, &req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := s.Proxy.Remove(req.Id); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.JSON(nil)
|
||||
}
|
||||
|
||||
// region 报告上线
|
||||
|
||||
@@ -797,10 +797,7 @@ type CreateResourceReq struct {
|
||||
// ResourcePrice 套餐价格
|
||||
func ResourcePrice(c *fiber.Ctx) error {
|
||||
// 检查权限
|
||||
_, err := auth.GetAuthCtx(c).PermitSecretClient()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ac := auth.GetAuthCtx(c)
|
||||
|
||||
// 解析请求参数
|
||||
var req = new(CreateResourceReq)
|
||||
@@ -809,7 +806,7 @@ func ResourcePrice(c *fiber.Ctx) error {
|
||||
}
|
||||
|
||||
// 获取套餐价格
|
||||
detail, err := req.TradeDetail(nil)
|
||||
detail, err := req.TradeDetail(ac.User)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -25,7 +25,6 @@ func ApplyRouters(app *fiber.App) {
|
||||
if env.RunMode == env.RunModeDev {
|
||||
debug := app.Group("/debug")
|
||||
debug.Get("/sms/:phone", handlers.DebugGetSmsCode)
|
||||
debug.Get("/proxy/register", handlers.DebugRegisterProxyBaiYin)
|
||||
debug.Get("/iden/clear/:phone", handlers.DebugIdentifyClear)
|
||||
debug.Get("/session/now", func(ctx *fiber.Ctx) error {
|
||||
rs, err := q.Session.Where(q.Session.AccessTokenExpires.Gt(time.Now())).Find()
|
||||
@@ -134,9 +133,6 @@ func clientRouter(api fiber.Router) {
|
||||
channel := client.Group("/channel")
|
||||
channel.Post("/remove", handlers.RemoveChannels)
|
||||
|
||||
// 代理网关注册
|
||||
proxy := client.Group("/proxy")
|
||||
proxy.Post("/register/baidyin", handlers.ProxyRegisterBaiYin)
|
||||
}
|
||||
|
||||
// 管理员接口路由
|
||||
@@ -196,6 +192,15 @@ func adminRouter(api fiber.Router) {
|
||||
channel.Post("/page", handlers.PageChannelByAdmin)
|
||||
channel.Post("/page/of-user", handlers.PageChannelOfUserByAdmin)
|
||||
|
||||
// proxy 代理
|
||||
var proxy = api.Group("/proxy")
|
||||
proxy.Post("/all", handlers.AllProxyByAdmin)
|
||||
proxy.Post("/page", handlers.PageProxyByAdmin)
|
||||
proxy.Post("/create", handlers.CreateProxy)
|
||||
proxy.Post("/update", handlers.UpdateProxy)
|
||||
proxy.Post("/update/status", handlers.UpdateProxyStatus)
|
||||
proxy.Post("/remove", handlers.RemoveProxy)
|
||||
|
||||
// trade 交易
|
||||
var trade = api.Group("/trade")
|
||||
trade.Post("/page", handlers.PageTradeByAdmin)
|
||||
|
||||
@@ -24,7 +24,7 @@ var Channel = &channelServer{
|
||||
}
|
||||
|
||||
type ChannelServiceProvider interface {
|
||||
CreateChannels(source netip.Addr, resourceId int32, authWhitelist bool, authPassword bool, count int, edgeFilter ...EdgeFilter) ([]*m.Channel, error)
|
||||
CreateChannels(source netip.Addr, resourceId int32, authWhitelist bool, authPassword bool, count int, edgeFilter *EdgeFilter) ([]*m.Channel, error)
|
||||
RemoveChannels(batch string) error
|
||||
}
|
||||
|
||||
@@ -32,8 +32,8 @@ type channelServer struct {
|
||||
provider ChannelServiceProvider
|
||||
}
|
||||
|
||||
func (s *channelServer) CreateChannels(source netip.Addr, resourceId int32, authWhitelist bool, authPassword bool, count int, edgeFilter ...EdgeFilter) ([]*m.Channel, error) {
|
||||
return s.provider.CreateChannels(source, resourceId, authWhitelist, authPassword, count, edgeFilter...)
|
||||
func (s *channelServer) CreateChannels(source netip.Addr, resourceId int32, authWhitelist bool, authPassword bool, count int, edgeFilter *EdgeFilter) ([]*m.Channel, error) {
|
||||
return s.provider.CreateChannels(source, resourceId, authWhitelist, authPassword, count, edgeFilter)
|
||||
}
|
||||
|
||||
func (s *channelServer) RemoveChannels(batch string) error {
|
||||
@@ -232,7 +232,7 @@ func regChans(proxy int32, chans []netip.AddrPort) error {
|
||||
// 缩容通道
|
||||
func remChans(proxy int32) error {
|
||||
key := freeChansKey + ":" + strconv.Itoa(int(proxy))
|
||||
err := g.Redis.SRem(context.Background(), key).Err()
|
||||
err := g.Redis.Del(context.Background(), key).Err()
|
||||
if err != nil {
|
||||
return fmt.Errorf("缩容通道失败: %w", err)
|
||||
}
|
||||
@@ -268,11 +268,12 @@ func lockChans(proxy int32, batch string, count int) ([]netip.AddrPort, error) {
|
||||
}
|
||||
|
||||
var RedisScriptLockChans = redis.NewScript(`
|
||||
local free_key = KEYS[1]
|
||||
local free_key = KEYS[1]
|
||||
local batch_key = KEYS[2]
|
||||
local count = tonumber(ARGV[1])
|
||||
|
||||
if redis.call("SCARD", free_key) < count then
|
||||
local free_count = redis.call("SCARD", free_key)
|
||||
if count <= 0 or free_count < count then
|
||||
return nil
|
||||
end
|
||||
|
||||
@@ -301,16 +302,17 @@ func freeChans(proxy int32, batch string) error {
|
||||
}
|
||||
|
||||
var RedisScriptFreeChans = redis.NewScript(`
|
||||
local free_key = KEYS[1]
|
||||
local free_key = KEYS[1]
|
||||
local batch_key = KEYS[2]
|
||||
|
||||
local chans = redis.call("LRANGE", batch_key, 0, -1)
|
||||
redis.call("DEL", batch_key)
|
||||
|
||||
if redis.call("EXISTS", free_key) == 1 then
|
||||
redis.call("SADD", free_key, unpack(chans))
|
||||
if #chans == 0 then
|
||||
return 1
|
||||
end
|
||||
|
||||
redis.call("SADD", free_key, unpack(chans))
|
||||
redis.call("DEL", batch_key)
|
||||
|
||||
return 1
|
||||
`)
|
||||
|
||||
|
||||
@@ -23,10 +23,9 @@ import (
|
||||
|
||||
type channelBaiyinProvider struct{}
|
||||
|
||||
func (s *channelBaiyinProvider) CreateChannels(source netip.Addr, resourceId int32, authWhitelist bool, authPassword bool, count int, edgeFilter ...EdgeFilter) ([]*m.Channel, error) {
|
||||
var filter *EdgeFilter = nil
|
||||
if len(edgeFilter) > 0 {
|
||||
filter = &edgeFilter[0]
|
||||
func (s *channelBaiyinProvider) CreateChannels(source netip.Addr, resourceId int32, authWhitelist bool, authPassword bool, count int, filter *EdgeFilter) ([]*m.Channel, error) {
|
||||
if filter == nil {
|
||||
return nil, core.NewBizErr("缺少节点过滤条件")
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
@@ -64,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
|
||||
}
|
||||
|
||||
// 获取可用节点
|
||||
@@ -86,7 +102,7 @@ func (s *channelBaiyinProvider) CreateChannels(source netip.Addr, resourceId int
|
||||
return nil, core.NewBizErr("获取可用节点失败", err)
|
||||
}
|
||||
if edgesResp.Total != count && len(edgesResp.Edges) != count {
|
||||
return nil, core.NewBizErr("地区可用节点数量不足 [%s, %s] [%s]")
|
||||
return nil, core.NewBizErr("地区可用节点数量不足")
|
||||
}
|
||||
edges := edgesResp.Edges
|
||||
|
||||
@@ -98,10 +114,6 @@ func (s *channelBaiyinProvider) CreateChannels(source netip.Addr, resourceId int
|
||||
ch := chans[i]
|
||||
edge := edges[i]
|
||||
|
||||
if err != nil {
|
||||
return nil, core.NewBizErr("解析通道地址失败", err)
|
||||
}
|
||||
|
||||
// 通道数据
|
||||
channels[i] = &m.Channel{
|
||||
UserID: user.ID,
|
||||
@@ -301,7 +313,7 @@ func (s *channelBaiyinProvider) RemoveChannels(batch string) error {
|
||||
})
|
||||
|
||||
// 清空通道配置
|
||||
secret := strings.Split(*proxy.Secret, ":")
|
||||
secret := strings.Split(u.Z(proxy.Secret), ":")
|
||||
gateway := g.NewGateway(proxy.IP.String(), secret[0], secret[1])
|
||||
err := gateway.GatewayPortConfigs(configs)
|
||||
if err != nil {
|
||||
@@ -321,6 +333,6 @@ func (s *channelBaiyinProvider) RemoveChannels(batch string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
slog.Debug("清除代理端口配置", "time", time.Since(start).String())
|
||||
slog.Debug("清除代理端口配置", "duration", time.Since(start).String())
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -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
|
||||
})
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package services
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"platform/pkg/u"
|
||||
"platform/web/core"
|
||||
m "platform/web/models"
|
||||
@@ -151,7 +152,8 @@ func (s *resourceService) CalcPrice(skuCode string, count int32, user *m.User, c
|
||||
Where(q.ProductSku.Code.Eq(skuCode), q.ProductSku.Status.Eq(int32(m.SkuStatusEnabled))).
|
||||
Take()
|
||||
if err != nil {
|
||||
return nil, nil, nil, decimal.Zero, decimal.Zero, decimal.Zero, core.NewServErr(fmt.Sprintf("产品不可用 %s", skuCode), err)
|
||||
slog.Debug("查询产品失败", "skuCode", skuCode)
|
||||
return nil, nil, nil, decimal.Zero, decimal.Zero, decimal.Zero, core.NewBizErr(fmt.Sprintf("产品不可用", skuCode), err)
|
||||
}
|
||||
|
||||
// 原价
|
||||
@@ -172,7 +174,7 @@ func (s *resourceService) CalcPrice(skuCode string, count int32, user *m.User, c
|
||||
}
|
||||
|
||||
uDiscountRate := uDiscount.Rate()
|
||||
if uDiscountRate.Cmp(discountRate) > 0 {
|
||||
if uDiscountRate.Cmp(discountRate) < 0 {
|
||||
discountRate = uDiscountRate
|
||||
discount = uDiscount
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user