diff --git a/common/busi_error.go b/common/busi_error.go index 58855a7..36d4380 100644 --- a/common/busi_error.go +++ b/common/busi_error.go @@ -61,6 +61,7 @@ const ( LiveProviderDisabled = 57004 LiveProviderInvalid = 57005 LiveLeaseNotFound = 57006 + LiveStopping = 57007 // 计费 58xxx TrafficNotEnough = 58001 @@ -122,6 +123,7 @@ var ( ErrLiveProviderDisabled = &BusiError{Code: LiveProviderDisabled, Msg: "直播服务未配置"} ErrLiveProviderInvalid = &BusiError{Code: LiveProviderInvalid, Msg: "直播服务配置无效"} ErrLiveLeaseNotFound = &BusiError{Code: LiveLeaseNotFound, Msg: "直播观看租约不存在或已过期"} + ErrLiveStopping = &BusiError{Code: LiveStopping, Msg: "上一场直播正在停止，请稍后重试"} ErrVideoNotFound = &BusiError{Code: VideoNotFound, Msg: "视频不存在"} ErrVideoUploading = &BusiError{Code: VideoUploading, Msg: "视频上传中"} diff --git a/common/redis.go b/common/redis.go index 6a0ea0b..de0c147 100644 --- a/common/redis.go +++ b/common/redis.go @@ -2,6 +2,8 @@ package common import ( "encoding/json" + "fmt" + "strings" "time" "laic-backend/logger" @@ -107,8 +109,27 @@ func SetRemoveIfKeyMissing(setKey, presenceKey, member string) (bool, error) { func SetNX(key string, value any) (bool, error) { c := _redis.Get() defer c.Close() - n, err := redis.Int(c.Do("SET", key, value, "NX")) - return n == 1, err + reply, err := c.Do("SET", key, value, "NX") + return setNXResult(reply, err) +} + +// SET ... NX returns the simple string "OK" when it writes and nil when the +// key already exists. It is not an integer reply. +func setNXResult(reply any, err error) (bool, error) { + if err != nil { + return false, err + } + if reply == nil { + return false, nil + } + status, err := redis.String(reply, nil) + if err != nil { + return false, err + } + if strings.EqualFold(status, "OK") { + return true, nil + } + return false, fmt.Errorf("redis: unexpected SET NX reply %q", status) } // IncrBy 原子自增，返回自增后的值（计费充值用） diff --git a/handler/payment_handler.go b/handler/payment_handler.go index dc8ce3e..058f72d 100644 --- a/handler/payment_handler.go +++ b/handler/payment_handler.go @@ -3,19 +3,23 @@ package handler import ( "bytes" "encoding/json" + "fmt" "io" "mime" "net/http" + "strings" "github.com/gin-gonic/gin" - "github.com/gin-gonic/gin/binding" "laic-backend/common" + "laic-backend/logger" "laic-backend/model" "laic-backend/service" "laic-backend/vo" ) +const maxAGPayCallbackBodyBytes = 64 * 1024 + func CreateWechatPayment(c *gin.Context) { var req vo.PaymentCreateReq if err := c.ShouldBindJSON(&req); err != nil { @@ -45,29 +49,21 @@ func GetPaymentTransaction(c *gin.Context) { } func AGPayCallback(c *gin.Context) { - contentType, _, err := mime.ParseMediaType(c.GetHeader("Content-Type")) - if err != nil || contentType != "application/json" { - c.Status(http.StatusBadRequest) - return - } - c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 1<<10) + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, maxAGPayCallbackBodyBytes) body, err := c.GetRawData() if err != nil { + logger.WARN("拒绝 AGPay 支付回调", "stage", "read_body", "error", err) c.Status(http.StatusBadRequest) return } - var callback model.AGPayCallback - decoder := json.NewDecoder(bytes.NewReader(body)) - decoder.DisallowUnknownFields() - if err := decoder.Decode(&callback); err != nil || decoder.Decode(&struct{}{}) != io.EOF { - c.Status(http.StatusBadRequest) - return - } - if err := binding.Validator.ValidateStruct(&callback); err != nil { + callback, err := parseAGPayCallback(c.GetHeader("Content-Type"), body) + if err != nil { + logger.WARN("拒绝 AGPay 支付回调", "stage", "parse", "contentType", c.GetHeader("Content-Type"), "bodyBytes", len(body), "error", err) c.Status(http.StatusBadRequest) return } if err := service.DefaultPaymentService.ConfirmAGPayCallback(callback.TradeNo, body); err != nil { + logger.WARN("拒绝 AGPay 支付回调", "stage", "confirm", "tradeNo", callback.TradeNo, "code", err.Code, "error", err) if err == common.ErrInternal { c.Status(http.StatusInternalServerError) return @@ -77,3 +73,29 @@ func AGPayCallback(c *gin.Context) { } c.JSON(http.StatusOK, gin.H{"code": "SUCCESS"}) } + +func parseAGPayCallback(contentType string, body []byte) (model.AGPayCallback, error) { + var callback model.AGPayCallback + mediaType, _, err := mime.ParseMediaType(contentType) + if err != nil || (mediaType != "application/json" && !strings.HasSuffix(mediaType, "+json")) { + return callback, fmt.Errorf("unsupported content type %q", contentType) + } + decoder := json.NewDecoder(bytes.NewReader(body)) + // Gateways commonly add payment amount, provider transaction ID, and + // signature fields. Only the merchant trade number is needed here; reject + // malformed payloads, but do not reject otherwise valid gateway metadata. + if err := decoder.Decode(&callback); err != nil { + return callback, err + } + if err := decoder.Decode(&struct{}{}); err != io.EOF { + if err == nil { + return callback, fmt.Errorf("multiple JSON values") + } + return callback, err + } + callback.TradeNo = strings.TrimSpace(callback.TradeNo) + if callback.TradeNo == "" || len(callback.TradeNo) > 128 { + return callback, fmt.Errorf("invalid tradeNo") + } + return callback, nil +} diff --git a/model/route.go b/model/route.go index 816a691..bac1bb6 100644 --- a/model/route.go +++ b/model/route.go @@ -4,12 +4,15 @@ import "time" // Route 航线 type Route struct { - ID int64 `gorm:"primaryKey;column:id;type:BIGINT;not null" json:"id"` - UserID int64 `gorm:"column:user_id;type:BIGINT;not null" json:"userId"` - Name string `gorm:"column:name;type:VARCHAR(128);not null" json:"name"` - Description string `gorm:"column:description;type:VARCHAR(256)" json:"description"` - CreatedAt time.Time `gorm:"column:created_at" json:"createdAt"` - UpdatedAt time.Time `gorm:"column:updated_at" json:"updatedAt"` + ID int64 `gorm:"primaryKey;column:id;type:BIGINT;not null" json:"id"` + UserID int64 `gorm:"column:user_id;type:BIGINT;not null" json:"userId"` + Name string `gorm:"column:name;type:VARCHAR(128);not null" json:"name"` + Description string `gorm:"column:description;type:VARCHAR(256)" json:"description"` + // WaypointCount is populated by route list/detail queries and is not stored + // on the route table. + WaypointCount int64 `gorm:"column:waypoint_count;->" json:"waypointCount"` + CreatedAt time.Time `gorm:"column:created_at" json:"createdAt"` + UpdatedAt time.Time `gorm:"column:updated_at" json:"updatedAt"` } func (Route) TableName() string { diff --git a/service/billing_service.go b/service/billing_service.go index d2e3531..8cb7609 100644 --- a/service/billing_service.go +++ b/service/billing_service.go @@ -321,12 +321,14 @@ func (b *BillingService) CreateOrderWithPackage(userID, packageID int64, amountG return err } var pending model.TrafficOrder - if err := tx.Where("user_id = ? AND pay_status IN ?", userID, []string{"unpaid", "processing"}).First(&pending).Error; err == nil { + pendingResult := tx.Where("user_id = ? AND pay_status IN ?", userID, []string{"unpaid", "processing"}).Limit(1).Find(&pending) + if pendingResult.Error != nil { + return pendingResult.Error + } + if pendingResult.RowsAffected > 0 { logger.WARN("创建流量订单被待支付订单阻止", "userId", userID, "pendingOrderId", pending.ID, "payStatus", pending.PayStatus) busiErr = common.ErrPendingOrderExists return busiErr - } else if !errors.Is(err, gorm.ErrRecordNotFound) { - return err } if packageID > 0 { var pkg model.TrafficPackage diff --git a/service/live_service.go b/service/live_service.go index 6da1845..5bdd96d 100644 --- a/service/live_service.go +++ b/service/live_service.go @@ -26,6 +26,8 @@ type LiveService struct { var DefaultLiveService = &LiveService{} +var errLiveStopping = errors.New("live session is stopping") + func pushURLHash(pushURL string) string { sum := sha256.Sum256([]byte(pushURL)) return hex.EncodeToString(sum[:]) @@ -73,11 +75,22 @@ func (s *LiveService) Join(userID int64, isAdmin bool, dockID string, req *vo.Li } // Find is intentional here: no active session is the normal first-join // path, so it should not be logged by GORM as a record-not-found error. - if result := tx.Where("dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", dockID). + if result := tx.Where("dock_id = ? AND phase IN ('starting','streaming','reconnecting')", dockID). Order("created_at DESC, id DESC").Limit(1).Find(&session); result.Error != nil { return result.Error } if session.ID == "" { + // A stopping session cannot be safely reused: its stop command is + // already in flight and the edge may transition it to stopped at any + // moment. Ask the caller to retry after that transition instead of + // handing back a session that will immediately become stopped. + var stopping model.LiveSession + if result := tx.Where("dock_id = ? AND phase = 'stopping'", dockID). + Order("created_at DESC, id DESC").Limit(1).Find(&stopping); result.Error != nil { + return result.Error + } else if result.RowsAffected > 0 { + return errLiveStopping + } maxBitrate := req.MaxBitrateBps if maxBitrate <= 0 { maxBitrate = 1500000 @@ -119,6 +132,9 @@ func (s *LiveService) Join(userID int64, isAdmin bool, dockID string, req *vo.Li }).Create(&lease).Error }) if err != nil { + if errors.Is(err, errLiveStopping) { + return nil, common.ErrLiveStopping + } logger.ERROR("创建直播观看租约失败", err) return nil, common.ErrInternal } diff --git a/service/payment_service.go b/service/payment_service.go index e4f557c..252aea1 100644 --- a/service/payment_service.go +++ b/service/payment_service.go @@ -175,12 +175,14 @@ func (s *PaymentService) getOrCreateTransaction(userID int64, businessType strin if err != nil { return err } - if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + existingResult := tx.Clauses(clause.Locking{Strength: "UPDATE"}). Where("user_id = ? AND business_type = ? AND business_order_id = ? AND provider = ?", userID, businessType, businessOrderID, "agpay"). - First(&transaction).Error; err == nil { + Limit(1).Find(&transaction) + if existingResult.Error != nil { + return existingResult.Error + } + if existingResult.RowsAffected > 0 { return nil - } else if !errors.Is(err, gorm.ErrRecordNotFound) { - return err } if amount <= 0 { busiErr = common.ErrParam @@ -303,8 +305,11 @@ func (s *PaymentService) ConfirmAGPayCallback(tradeNo string, payload []byte) *c } var event model.PaymentCallbackEvent - err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("provider = ? AND event_key = ?", "agpay", tradeNo).First(&event).Error - if err == nil { + eventResult := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("provider = ? AND event_key = ?", "agpay", tradeNo).Limit(1).Find(&event) + if eventResult.Error != nil { + return eventResult.Error + } + if eventResult.RowsAffected > 0 { callbackTransaction = transaction if event.PayloadHash != payloadHash { callbackAudit = "conflict" @@ -315,9 +320,6 @@ func (s *PaymentService) ConfirmAGPayCallback(tradeNo string, payload []byte) *c callbackTransaction = transaction return nil } - if !errors.Is(err, gorm.ErrRecordNotFound) { - return err - } id, err := tool.NextID() if err != nil { return err diff --git a/service/route_service.go b/service/route_service.go index 5984e61..90c0c1a 100644 --- a/service/route_service.go +++ b/service/route_service.go @@ -31,6 +31,10 @@ func (s *RouteService) GetPage(userID int64, isAdmin bool, req *vo.RoutePageReq) return nil, common.ErrInternal } var list []model.Route + // Count in the same list query so the endpoint does not issue one waypoint + // query per route. The read-only Route.WaypointCount field receives the + // correlated subquery result. + db = db.Select("route.*, (SELECT COUNT(*) FROM route_waypoint rw WHERE rw.route_id = route.id) AS waypoint_count") if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询航线列表失败", err) return nil, common.ErrInternal @@ -52,6 +56,7 @@ func (s *RouteService) GetDetail(userID int64, isAdmin bool, id int64) (*vo.Rout logger.ERROR("查询航点失败", err) return nil, common.ErrInternal } + route.WaypointCount = int64(len(waypoints)) return &vo.RouteVO{Route: &route, Waypoints: waypoints}, nil } @@ -88,6 +93,7 @@ func (s *RouteService) Create(userID int64, req *vo.RouteCreateReq) (*vo.RouteVO logger.ERROR("新增航线失败", err) return nil, common.ErrInternal } + route.WaypointCount = int64(len(waypoints)) return &vo.RouteVO{Route: route, Waypoints: waypoints}, nil } @@ -133,6 +139,7 @@ func (s *RouteService) Update(userID int64, isAdmin bool, id int64, req *vo.Rout common.DB.First(&route, id) waypoints, _ := s.waypoints(id) + route.WaypointCount = int64(len(waypoints)) return &vo.RouteVO{Route: &route, Waypoints: waypoints}, nil }