Compare commits
3 Commits
v1.11.0
...
5bb65216a9
| Author | SHA1 | Date | |
|---|---|---|---|
| 5bb65216a9 | |||
| 9c6b07ad8c | |||
| 49d2df54f5 |
@@ -1,5 +1,10 @@
|
||||
## TODO
|
||||
|
||||
后台交易备注
|
||||
关闭支付弹窗
|
||||
兜底手动操作
|
||||
限速
|
||||
|
||||
- edge.area_id 可为空,代表节点无固定地区
|
||||
- 后台展示 mac, ip:port,实际地区
|
||||
|
||||
|
||||
@@ -40,14 +40,6 @@ services:
|
||||
depends_on:
|
||||
- redis
|
||||
|
||||
gost:
|
||||
image: gogost/gost
|
||||
command: >
|
||||
-api test:test@:9700
|
||||
ports:
|
||||
- "9700:9700"
|
||||
restart: unless-stopped
|
||||
|
||||
volumes:
|
||||
postgres_data:
|
||||
redis_data:
|
||||
|
||||
@@ -184,6 +184,7 @@ values
|
||||
('许昌', 2, (select id from area where name = '河南')),
|
||||
('郑州', 2, (select id from area where name = '河南')),
|
||||
('驻马店', 2, (select id from area where name = '河南')),
|
||||
('平顶山', 2, (select id from area where name = '河南')),
|
||||
('鹤壁', 2, (select id from area where name = '河南')),
|
||||
('丽水', 2, (select id from area where name = '浙江')),
|
||||
('台州', 2, (select id from area where name = '浙江')),
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -614,8 +614,8 @@ create table proxy (
|
||||
updated_at timestamptz default current_timestamp,
|
||||
deleted_at timestamptz
|
||||
);
|
||||
create unique index udx_proxy_mac on proxy (mac) where deleted_at is null;
|
||||
create unique index udx_proxy_ip on proxy (ip) where deleted_at is null;
|
||||
create index idx_proxy_mac on proxy (mac) where deleted_at is null;
|
||||
create index idx_proxy_ip on proxy (ip) where deleted_at is null;
|
||||
create index idx_proxy_created_at on proxy (created_at) where deleted_at is null;
|
||||
|
||||
-- proxy表字段注释
|
||||
|
||||
@@ -1,9 +1,7 @@
|
||||
package events
|
||||
|
||||
import "github.com/hibiken/asynq"
|
||||
|
||||
const RefreshEdge = "edge:refresh"
|
||||
|
||||
func NewRefreshEdge() *asynq.Task {
|
||||
return asynq.NewTask(RefreshEdge, nil)
|
||||
}
|
||||
// func NewRefreshEdge() *asynq.Task {
|
||||
// return asynq.NewTask(RefreshEdge, nil)
|
||||
// }
|
||||
|
||||
@@ -42,7 +42,7 @@ type gostClient struct {
|
||||
var GostInitializer = func(host string, port int, pathPrefix, username, password string) GostClient {
|
||||
baseURL := strings.TrimSpace(host)
|
||||
if !strings.Contains(baseURL, "://") {
|
||||
baseURL = fmt.Sprintf("http://%s:%d", baseURL, port)
|
||||
baseURL = fmt.Sprintf("https://%s", baseURL)
|
||||
}
|
||||
|
||||
return &gostClient{
|
||||
@@ -89,6 +89,7 @@ type GostServiceConfig struct {
|
||||
Handler GostHandlerConfig `json:"handler"`
|
||||
Listener GostListenerConfig `json:"listener"`
|
||||
Recorders []GostRecorderConfig `json:"recorders,omitempty"`
|
||||
Limiter string `json:"limiter,omitempty"`
|
||||
}
|
||||
|
||||
type GostHandlerConfig struct {
|
||||
|
||||
@@ -422,7 +422,7 @@ func PageResourceShortOfUserByAdmin(c *fiber.Ctx) error {
|
||||
do = do.Where(q.Resource.CreatedAt.Lte(req.CreatedAtEnd.UTC()))
|
||||
}
|
||||
|
||||
list, total, err := q.Resource.Debug().
|
||||
list, total, err := q.Resource.
|
||||
Joins(q.Resource.User, q.Resource.Short, q.Resource.Short.Sku).
|
||||
Select(
|
||||
q.Resource.ALL,
|
||||
|
||||
@@ -1,10 +1,7 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"platform/pkg/env"
|
||||
"platform/web/auth"
|
||||
"platform/web/core"
|
||||
g "platform/web/globals"
|
||||
@@ -14,7 +11,6 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/gofiber/fiber/v2"
|
||||
"github.com/valyala/fasthttp"
|
||||
)
|
||||
|
||||
// PageTradeByAdmin 分页查询所有订单
|
||||
@@ -221,6 +217,36 @@ type TradeCreateReq struct {
|
||||
|
||||
// ============================================================
|
||||
|
||||
// 更新订单备注
|
||||
func TradeUpdateRemarkByAdmin(c *fiber.Ctx) error {
|
||||
// 检查权限
|
||||
_, err := auth.GetAuthCtx(c).PermitAdmin()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 解析请求参数
|
||||
var req TradeUpdateRemarkReq
|
||||
if err := g.Validator.ParseBody(c, &req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 更新订单备注
|
||||
err = s.Trade.UpdateRemark(req.TradeNo, req.Remark)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.SendStatus(fiber.StatusNoContent)
|
||||
}
|
||||
|
||||
type TradeUpdateRemarkReq struct {
|
||||
TradeNo string `json:"trade_no" validate:"required"`
|
||||
Remark string `json:"remark"`
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
|
||||
// 完成订单
|
||||
func TradeComplete(c *fiber.Ctx) error {
|
||||
// 检查权限
|
||||
@@ -276,6 +302,29 @@ func TradeCompleteByAdmin(c *fiber.Ctx) error {
|
||||
return c.SendStatus(fiber.StatusNoContent)
|
||||
}
|
||||
|
||||
// 订单补余额
|
||||
func TradeConvertByAdmin(c *fiber.Ctx) error {
|
||||
// 检查权限
|
||||
authCtx, err := auth.GetAuthCtx(c).PermitAdmin(core.ScopeTradeWrite)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 解析请求参数
|
||||
var req s.TradeRef
|
||||
if err := g.Validator.ParseBody(c, &req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 订单补余额
|
||||
err = s.Trade.ConvertTradeToBalance(authCtx.Admin, &req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.SendStatus(fiber.StatusNoContent)
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
|
||||
// 取消订单
|
||||
@@ -287,14 +336,14 @@ func TradeCancel(c *fiber.Ctx) error {
|
||||
}
|
||||
|
||||
// 解析请求参数
|
||||
req := new(TradeCancelReq)
|
||||
req := new(s.TradeRef)
|
||||
if err := g.Validator.ParseBody(c, req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 取消交易
|
||||
err = s.Trade.CancelTrade(&req.TradeRef)
|
||||
if err != nil {
|
||||
err = s.Trade.CancelTrade(req)
|
||||
if err != nil && err != s.ErrTradeStatusIgnored {
|
||||
slog.Error("取消交易失败", "trade_no", req.TradeNo, "error", err)
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "取消交易失败"})
|
||||
}
|
||||
@@ -302,62 +351,63 @@ func TradeCancel(c *fiber.Ctx) error {
|
||||
return c.SendStatus(fiber.StatusNoContent)
|
||||
}
|
||||
|
||||
type TradeCancelReq struct {
|
||||
s.TradeRef
|
||||
// ============================================================
|
||||
|
||||
// 结束订单:完成或取消订单
|
||||
func TradeFinish(c *fiber.Ctx) error {
|
||||
// 检查权限
|
||||
authCtx, err := auth.GetAuthCtx(c).PermitUser()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 解析请求参数
|
||||
var req s.TradeRef
|
||||
if err := g.Validator.ParseBody(c, &req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 尝试取消交易
|
||||
err = s.Trade.CancelTrade(&req)
|
||||
if err == s.ErrTradeStatusIgnored {
|
||||
// 尝试完成交易
|
||||
err = s.Trade.CompleteTrade(authCtx.User, &req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.JSON(map[string]m.TradeStatus{
|
||||
"status": m.TradeStatusSuccess,
|
||||
})
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.JSON(map[string]m.TradeStatus{
|
||||
"status": m.TradeStatusCanceled,
|
||||
})
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
|
||||
// 检查订单
|
||||
func TradeCheck(c *fiber.Ctx) error {
|
||||
// 检查权限:sse 接口暂时不检查权限
|
||||
|
||||
// 解析请求参数
|
||||
req := new(TradeCheckReq)
|
||||
if err := g.Validator.ParseQuery(c, req); err != nil {
|
||||
_, err := auth.GetAuthCtx(c).PermitAdmin(core.ScopeTradeRead)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.Set(fiber.HeaderContentType, "text/event-stream")
|
||||
c.Set(fiber.HeaderCacheControl, "no-cache")
|
||||
c.Set(fiber.HeaderConnection, "keep-alive")
|
||||
c.Set(fiber.HeaderTransferEncoding, "chunked")
|
||||
c.Context().SetBodyStreamWriter(fasthttp.StreamWriter(func(w *bufio.Writer) {
|
||||
// 解析请求参数
|
||||
var req s.TradeRef
|
||||
if err := g.Validator.ParseBody(c, &req); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
expire := env.TradeExpire
|
||||
interval := 5
|
||||
for range expire / interval {
|
||||
// 检查订单状态
|
||||
result, err := s.Trade.CheckTrade(&req.TradeRef)
|
||||
if err != nil {
|
||||
slog.Error("检查订单状态失败", "trade_no", req.TradeNo, "error", err)
|
||||
return
|
||||
}
|
||||
// 检查订单状态
|
||||
result, err := s.Trade.CheckTrade(&req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 写入订单状态
|
||||
_, err = fmt.Fprintf(w, "data: %d\n\n", result.Status)
|
||||
if err != nil {
|
||||
slog.Error("写入订单状态失败", "trade_no", req.TradeNo, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
err = w.Flush()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// 当订单离开支付状态后结束查询
|
||||
if result.Status != m.TradeStatusPending {
|
||||
return
|
||||
}
|
||||
|
||||
time.Sleep(time.Duration(interval) * time.Second)
|
||||
}
|
||||
}))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type TradeCheckReq struct {
|
||||
s.TradeRef
|
||||
return c.JSON(result)
|
||||
}
|
||||
|
||||
@@ -66,10 +66,6 @@ func publicRouter(api fiber.Router) {
|
||||
resource := api.Group("/resource")
|
||||
resource.Post("/price", handlers.ResourcePrice)
|
||||
|
||||
// 交易
|
||||
trade := api.Group("/trade")
|
||||
trade.Get("/check", handlers.TradeCheck)
|
||||
|
||||
// 前台
|
||||
inquiry := api.Group("/inquiry")
|
||||
inquiry.Post("/create", handlers.CreateInquiry)
|
||||
@@ -144,6 +140,7 @@ func userRouter(api fiber.Router) {
|
||||
trade.Post("/create", handlers.TradeCreate)
|
||||
trade.Post("/complete", handlers.TradeComplete)
|
||||
trade.Post("/cancel", handlers.TradeCancel)
|
||||
trade.Post("/finish", handlers.TradeFinish)
|
||||
|
||||
// 账单
|
||||
bill := api.Group("/bill")
|
||||
@@ -241,6 +238,8 @@ func adminRouter(api fiber.Router) {
|
||||
trade.Post("/page", handlers.PageTradeByAdmin)
|
||||
trade.Post("/page/of-user", handlers.PageTradeOfUserByAdmin)
|
||||
trade.Post("/complete", handlers.TradeCompleteByAdmin)
|
||||
trade.Post("/update/remark", handlers.TradeUpdateRemarkByAdmin)
|
||||
trade.Post("/check", handlers.TradeCheck)
|
||||
|
||||
// bill 账单
|
||||
var bill = api.Group("/bill")
|
||||
|
||||
@@ -136,7 +136,12 @@ func (s *channelServer) RemoveChannels(batch string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := s.provider.removeRemote(batch, usedBatch); err != nil {
|
||||
provider, err := channelProviderByProxyID(usedBatch.ProxyID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := provider.removeRemote(batch, usedBatch); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := freeChans(usedBatch.ProxyID, batch); err != nil {
|
||||
@@ -407,6 +412,22 @@ func selectProxyByType(proxyType m.ProxyType, count int) (*m.Proxy, error) {
|
||||
return bestProxy, nil
|
||||
}
|
||||
|
||||
func channelProviderByProxyID(proxyID int32) (channelProvider, error) {
|
||||
proxy, err := q.Proxy.Where(q.Proxy.ID.Eq(proxyID)).Take()
|
||||
if err != nil {
|
||||
return nil, core.NewServErr("获取代理数据失败", err)
|
||||
}
|
||||
|
||||
switch proxy.Type {
|
||||
case m.ProxyTypeGost:
|
||||
return &channelGostProvider{}, nil
|
||||
case m.ProxyTypeBaiYin:
|
||||
return &channelBaiyinProvider{}, nil
|
||||
default:
|
||||
return nil, core.NewBizErr(fmt.Sprintf("不支持的代理类型: %d", proxy.Type))
|
||||
}
|
||||
}
|
||||
|
||||
func (s *channelServer) RefreshEdges() error {
|
||||
|
||||
// 仅白银网关支持边缘节点刷新,GOST 不参与此流程。
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"platform/web/core"
|
||||
g "platform/web/globals"
|
||||
m "platform/web/models"
|
||||
q "platform/web/queries"
|
||||
)
|
||||
|
||||
type channelBaiyinProvider struct{}
|
||||
@@ -68,30 +67,31 @@ func (s *channelBaiyinProvider) prepareCreate(ctx *channelCreateContext) (*chann
|
||||
}
|
||||
|
||||
func (s *channelBaiyinProvider) removeRemote(_ string, batch *usedChanBatch) error {
|
||||
configs := make([]*g.PortConfigsReq, len(batch.Chans))
|
||||
for i, ch := range batch.Chans {
|
||||
configs[i] = &g.PortConfigsReq{
|
||||
Port: int(ch.Port()),
|
||||
Edge: &[]string{},
|
||||
AutoEdgeConfig: &g.AutoEdgeConfig{Count: u.P(0)},
|
||||
Status: false,
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
if err = gateway.GatewayPortConfigs(configs); err != nil {
|
||||
return core.NewServErr(fmt.Sprintf("清空代理 %s 端口配置失败", proxy.IP.String()), err)
|
||||
}
|
||||
return nil
|
||||
// configs := make([]*g.PortConfigsReq, len(batch.Chans))
|
||||
// for i, ch := range batch.Chans {
|
||||
// configs[i] = &g.PortConfigsReq{
|
||||
// Port: int(ch.Port()),
|
||||
// Edge: &[]string{},
|
||||
// AutoEdgeConfig: &g.AutoEdgeConfig{Count: u.P(0)},
|
||||
// Status: false,
|
||||
// }
|
||||
// }
|
||||
|
||||
// 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)
|
||||
// }
|
||||
|
||||
// if err = gateway.GatewayPortConfigs(configs); err != nil {
|
||||
// return core.NewServErr(fmt.Sprintf("清空代理 %s 端口配置失败", proxy.IP.String()), err)
|
||||
// }
|
||||
// return nil
|
||||
}
|
||||
|
||||
// ensureEdges 检查本地节点是否足够,如果不足从云端连入
|
||||
|
||||
@@ -51,8 +51,9 @@ func (s *channelGostProvider) prepareCreate(ctx *channelCreateContext) (*channel
|
||||
Type: "tcp",
|
||||
},
|
||||
Recorders: []g.GostRecorderConfig{
|
||||
{Name: "record-file", Record: "recorder.service.handler"},
|
||||
{Name: "record-http-otel", Record: "recorder.service.handler"},
|
||||
},
|
||||
Limiter: "limiter-8m",
|
||||
}
|
||||
|
||||
if ctx.AuthWhitelist {
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
q "platform/web/queries"
|
||||
"time"
|
||||
|
||||
"github.com/shopspring/decimal"
|
||||
"gorm.io/gen/field"
|
||||
)
|
||||
|
||||
@@ -20,7 +21,7 @@ func (s *productService) AllProducts() ([]*m.Product, error) {
|
||||
Find()
|
||||
}
|
||||
|
||||
func (s *productService) AllProductSaleInfos() ([]*m.Product, error) {
|
||||
func (s *productService) AllProductSaleInfos() ([]any, error) {
|
||||
products, err := q.Product.
|
||||
Select(
|
||||
q.Product.ID,
|
||||
@@ -43,7 +44,17 @@ func (s *productService) AllProductSaleInfos() ([]*m.Product, error) {
|
||||
pids[i] = p.ID
|
||||
}
|
||||
|
||||
skus, err := q.ProductSku.
|
||||
type SkuInfo struct {
|
||||
ID int32 `json:"id"`
|
||||
ProductID int32 `json:"product_id"`
|
||||
Name string `json:"name"`
|
||||
Code string `json:"code"`
|
||||
Price decimal.Decimal `json:"price"`
|
||||
CountMin decimal.Decimal `json:"count_min"`
|
||||
Discount float64 `json:"discount"`
|
||||
}
|
||||
var skus []*SkuInfo
|
||||
err = q.ProductSku.
|
||||
Select(
|
||||
q.ProductSku.ID,
|
||||
q.ProductSku.ProductID,
|
||||
@@ -51,29 +62,47 @@ func (s *productService) AllProductSaleInfos() ([]*m.Product, error) {
|
||||
q.ProductSku.Code,
|
||||
q.ProductSku.Price,
|
||||
q.ProductSku.CountMin,
|
||||
q.ProductDiscount.Discount,
|
||||
).
|
||||
Where(
|
||||
q.ProductSku.ProductID.In(pids...),
|
||||
q.ProductSku.Status.Eq(int32(m.SkuStatusEnabled)),
|
||||
).
|
||||
LeftJoin(q.ProductDiscount, q.ProductDiscount.ID.EqCol(q.ProductSku.DiscountId)).
|
||||
Order(q.ProductSku.Sort).
|
||||
Find()
|
||||
Scan(&skus)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pmap := make(map[int32]*m.Product, len(products))
|
||||
type ProductInfo struct {
|
||||
m.Product
|
||||
Skus []*SkuInfo `json:"skus,omitempty"`
|
||||
}
|
||||
pmap := make(map[int32]*ProductInfo, len(products))
|
||||
for _, p := range products {
|
||||
pmap[p.ID] = p
|
||||
p.Skus = make([]*m.ProductSku, 0)
|
||||
pmap[p.ID] = &ProductInfo{Product: *p, Skus: make([]*SkuInfo, 0)}
|
||||
}
|
||||
for _, s := range skus {
|
||||
if p, ok := pmap[s.ProductID]; ok {
|
||||
p.Skus = append(p.Skus, s)
|
||||
p.Skus = append(p.Skus, &SkuInfo{
|
||||
ID: s.ID,
|
||||
ProductID: s.ProductID,
|
||||
Name: s.Name,
|
||||
Code: s.Code,
|
||||
Price: s.Price,
|
||||
CountMin: s.CountMin,
|
||||
Discount: s.Discount,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return products, nil
|
||||
plist := make([]any, 0, len(pmap))
|
||||
for _, p := range pmap {
|
||||
plist = append(plist, p)
|
||||
}
|
||||
|
||||
return plist, nil
|
||||
}
|
||||
|
||||
// 新增产品
|
||||
|
||||
@@ -204,12 +204,7 @@ func (s *tradeService) Create(user *m.User, tradeData *CreateTradeData, productD
|
||||
// 缓存产品数据
|
||||
w := bytes.Buffer{}
|
||||
gob.NewEncoder(&w).Encode(detail)
|
||||
err = g.Redis.Set(
|
||||
context.Background(),
|
||||
tradeProductKey(tradeNo),
|
||||
w.Bytes(),
|
||||
expireIn,
|
||||
).Err()
|
||||
err = g.Redis.Set(context.Background(), tradeProductKey(tradeNo), w.Bytes(), 0).Err()
|
||||
if err != nil {
|
||||
return nil, core.NewServErr("保存购买信息失败", err)
|
||||
}
|
||||
@@ -271,7 +266,8 @@ func (s *tradeService) OnCompleteTrade(user *m.User, interNo string, outerNo str
|
||||
}
|
||||
|
||||
// 恢复购买信息;如果反序列化失败,检查开头 init 函数中是否注册了对应的 struct 类型
|
||||
detailBytes, err := g.Redis.Get(context.Background(), tradeProductKey(interNo)).Bytes()
|
||||
tradeKey := tradeProductKey(interNo)
|
||||
detailBytes, err := g.Redis.Get(context.Background(), tradeKey).Bytes()
|
||||
if err != nil {
|
||||
return core.NewServErr("恢复购买信息失败", err)
|
||||
}
|
||||
@@ -344,6 +340,12 @@ func (s *tradeService) OnCompleteTrade(user *m.User, interNo string, outerNo str
|
||||
}
|
||||
}
|
||||
|
||||
// 删除缓存
|
||||
err = g.Redis.Del(context.Background(), tradeKey).Err()
|
||||
if err != nil {
|
||||
return core.NewServErr("删除缓存失败", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
@@ -353,6 +355,23 @@ func (s *tradeService) OnCompleteTrade(user *m.User, interNo string, outerNo str
|
||||
return nil
|
||||
}
|
||||
|
||||
// 转换交易
|
||||
func (s *tradeService) ConvertTradeToBalance(admin *m.Admin, ref *TradeRef) error {
|
||||
trade, err := q.Trade.Where(q.Trade.InnerNo.Eq(ref.TradeNo)).First()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
user, err := q.User.Where(q.User.ID.Eq(trade.UserID)).First()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return q.Q.Transaction(func(q *q.Query) error {
|
||||
return User.UpdateBalance(q, user, trade.Payment, "管理员订单补余额", &admin.ID, nil)
|
||||
})
|
||||
}
|
||||
|
||||
// 取消交易
|
||||
func (s *tradeService) CancelTrade(ref *TradeRef) error {
|
||||
now := time.Now()
|
||||
@@ -394,7 +413,7 @@ func (s *tradeService) CancelTrade(ref *TradeRef) error {
|
||||
})
|
||||
if err != nil {
|
||||
slog.Debug(fmt.Sprintf("订单无需关闭: %s", err.Error()))
|
||||
return nil
|
||||
return ErrTradeStatusIgnored
|
||||
}
|
||||
|
||||
default:
|
||||
@@ -575,6 +594,12 @@ func (s *tradeService) CheckTrade(ref *TradeRef) (*CheckTradeResult, error) {
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
// 更新备注
|
||||
func (s *tradeService) UpdateRemark(tradeNo string, remark string) error {
|
||||
_, err := q.Trade.Where(q.Trade.InnerNo.Eq(tradeNo)).UpdateColumn(q.Trade.Remark, remark)
|
||||
return err
|
||||
}
|
||||
|
||||
func tradeProductKey(no string) string {
|
||||
return fmt.Sprintf("trade:%s:product", no)
|
||||
}
|
||||
@@ -639,4 +664,5 @@ func (e TradeErr) Error() string {
|
||||
|
||||
var (
|
||||
ErrTransactionNotSupported = core.NewBizErr("不支持的支付方式")
|
||||
ErrTradeStatusIgnored = core.NewBizErr("交易状态已忽略")
|
||||
)
|
||||
|
||||
@@ -3,10 +3,11 @@ package tasks
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"platform/web/events"
|
||||
q "platform/web/queries"
|
||||
m "platform/web/models"
|
||||
s "platform/web/services"
|
||||
|
||||
"github.com/hibiken/asynq"
|
||||
@@ -24,19 +25,30 @@ func HandleCompleteTrade(_ context.Context, task *asynq.Task) error {
|
||||
Method: event.Method,
|
||||
}
|
||||
|
||||
// 尝试完成交易
|
||||
user, err := s.User.Get(q.Q, event.UserId)
|
||||
if err != nil {
|
||||
return fmt.Errorf("获取用户失败: %w", err)
|
||||
// 关闭交易
|
||||
err := s.Trade.CancelTrade(&data)
|
||||
if errors.Is(err, s.ErrTradeStatusIgnored) {
|
||||
result, err := s.Trade.CheckTrade(&data)
|
||||
if err != nil {
|
||||
return fmt.Errorf("检查交易状态失败: %w", err)
|
||||
}
|
||||
|
||||
switch result.Status {
|
||||
case m.TradeStatusSuccess:
|
||||
if err := s.Trade.UpdateRemark(data.TradeNo, "已付款"); err != nil {
|
||||
slog.Error("添加备注失败", "err", err)
|
||||
}
|
||||
case m.TradeStatusCanceled:
|
||||
slog.Debug("交易已取消", "status", result.Status)
|
||||
default:
|
||||
return fmt.Errorf("意外交易状态: %v", result.Status)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := s.Trade.CompleteTrade(user, &data); err != nil {
|
||||
slog.Debug("结束交易失败:完成交易失败", "err", err)
|
||||
|
||||
// 交易无法完成,关闭交易
|
||||
if err := s.Trade.CancelTrade(&data); err != nil {
|
||||
return fmt.Errorf("结束交易失败:取消交易失败: %w", err)
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("结束交易失败:取消交易失败: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
@@ -56,9 +68,9 @@ func HandleRemoveChannel(_ context.Context, task *asynq.Task) (err error) {
|
||||
func HandleRefreshEdges(_ context.Context, task *asynq.Task) (err error) {
|
||||
slog.Info("[event]刷新边缘节点")
|
||||
|
||||
err = s.Channel.RefreshEdges()
|
||||
if err != nil {
|
||||
return fmt.Errorf("刷新边缘节点失败: %w", err)
|
||||
}
|
||||
// err = s.Channel.RefreshEdges()
|
||||
// if err != nil {
|
||||
// return fmt.Errorf("刷新边缘节点失败: %w", err)
|
||||
// }
|
||||
return nil
|
||||
}
|
||||
|
||||
27
web/web.go
27
web/web.go
@@ -42,10 +42,6 @@ func RunApp(pCtx context.Context) error {
|
||||
return RunTask(ctx)
|
||||
})
|
||||
|
||||
g.Go(func() error {
|
||||
return RunCron(ctx)
|
||||
})
|
||||
|
||||
return g.Wait()
|
||||
}
|
||||
|
||||
@@ -111,29 +107,6 @@ func RunTask(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func RunCron(ctx context.Context) error {
|
||||
cron := asynq.NewSchedulerFromRedisClient(deps.Redis, &asynq.SchedulerOpts{
|
||||
Logger: &AppAsynqLogger{},
|
||||
Location: time.Local,
|
||||
})
|
||||
|
||||
cron.Register("0/10 * * * *", events.NewRefreshEdge())
|
||||
|
||||
// 停止服务
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
cron.Shutdown()
|
||||
}()
|
||||
|
||||
// 启动服务
|
||||
err := cron.Run()
|
||||
if err != nil {
|
||||
return fmt.Errorf("定时任务服务运行失败: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type AppAsynqLogger struct{}
|
||||
|
||||
func (l *AppAsynqLogger) Debug(args ...any) {
|
||||
|
||||
Reference in New Issue
Block a user