From 76ccd5e50afffc510ef2f7c4328809d3abe66d6d Mon Sep 17 00:00:00 2001 From: liuhaodong Date: Wed, 2 Sep 2026 11:14:26 +0800 Subject: [PATCH] =?UTF-8?q?feat=EF=BC=9A=E5=8A=9F=E8=83=BD=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 39 +++ common/config.go | 7 + config-prod.yaml | 2 + config.yaml | 2 + handler/live_handler.go | 4 +- handler/platform_billing_handler.go | 21 ++ mock/dock.go | 377 ++++++++++++++++++++++++---- mock/main.go | 35 ++- mock/mission.go | 19 +- model/billing.go | 2 +- route/route.go | 1 + service/billing_service.go | 78 +++++- service/live_service.go | 58 ++++- service/payment_service.go | 51 +++- sql/001_schema.sql | 1 + vo/platform_billing_vo.go | 8 + 16 files changed, 633 insertions(+), 72 deletions(-) diff --git a/README.md b/README.md index 916a860..9112f1e 100644 --- a/README.md +++ b/README.md @@ -69,3 +69,42 @@ sudo tail -f /usr/local/laic-backend-prod/log/err.log ```bash sudo supervisorctl restart laic-backend-prod ``` + +## Payment and SIM schema reconciliation + +If a production database was upgraded from an older partial release, back it +up first and then run `sql/019_reconcile_payment_schema.sql`. The migration is +repeatable and fills missing `traffic_order.updated_at`, payment fields, pending +order guards, and normalizes legacy empty payment idempotency keys. It also +creates the payment/SIM tables. It does not create orders, charge users, or +create SIM cards. Verify the schema and platform resource pool after it finishes, then +restart the backend before retrying payment. + +AGPay returns the WeChat QR code as binary `image/png`. The backend converts it +to a JSON-safe `providerPayload` data URI (`data:image/png;base64,...`); clients +should use that value directly as the image `src` and must not UTF-8 decode the +original response body. + +## Test traffic grant + +For QA environments, set the following configuration explicitly (it defaults +to disabled): + +```yaml +testing: + allow-traffic-grant: true +``` + +An authenticated administrator can then credit a regular user without creating +a payment order: + +```http +POST /v1/admin/traffic/test-grants +Content-Type: application/json + +{"userId":8243938201600,"amountGb":10,"reason":"live test"} +``` + +The operation updates the user snapshot, appends a `test_grant` traffic ledger +entry, invalidates the Redis balance cache, and writes an operation log. Keep +the switch disabled in production after testing. diff --git a/common/config.go b/common/config.go index 74de4e8..33a2a7d 100644 --- a/common/config.go +++ b/common/config.go @@ -20,6 +20,7 @@ type AppConfig struct { Simboss Simboss `mapstructure:"simboss"` AGPay AGPay `mapstructure:"agpay"` Heartbeat Heartbeat `mapstructure:"heartbeat"` + Testing Testing `mapstructure:"testing"` } // AppConf 全局配置（LoadConfig 后可用） @@ -106,6 +107,12 @@ type Heartbeat struct { ScanSeconds int `mapstructure:"scan-seconds"` } +// Testing contains explicitly opt-in helpers that are only appropriate for +// non-production environments. +type Testing struct { + AllowTrafficGrant bool `mapstructure:"allow-traffic-grant"` +} + // LoadConfig 使用 Viper 加载本地 config.yaml func LoadConfig(path string, conf *AppConfig) { viper.SetConfigFile(path) diff --git a/config-prod.yaml b/config-prod.yaml index f8fa885..101a953 100644 --- a/config-prod.yaml +++ b/config-prod.yaml @@ -57,6 +57,8 @@ agpay: heartbeat: timeout-seconds: 30 scan-seconds: 5 +testing: + allow-traffic-grant: false log: level: info path: ./logs/ diff --git a/config.yaml b/config.yaml index 6ebc275..2e5cc5d 100644 --- a/config.yaml +++ b/config.yaml @@ -58,6 +58,8 @@ agpay: heartbeat: timeout-seconds: 30 scan-seconds: 5 +testing: + allow-traffic-grant: false log: level: debug path: ./logs/ diff --git a/handler/live_handler.go b/handler/live_handler.go index 368997a..5a77d39 100644 --- a/handler/live_handler.go +++ b/handler/live_handler.go @@ -129,7 +129,9 @@ func StopLive(c *gin.Context) { // GetLivePlayURL 获取直播播放地址 func GetLivePlayURL(c *gin.Context) { dockID, sessionID := c.Param("dockId"), c.Query("streamSessionId") - if dockID == "" || sessionID == "" { + // streamSessionId is optional for compatibility with older web clients. The + // service resolves it from the caller's active viewer lease when omitted. + if dockID == "" { common.FailWithBusiError(c, common.ErrParam) return } diff --git a/handler/platform_billing_handler.go b/handler/platform_billing_handler.go index dc80fea..685e064 100644 --- a/handler/platform_billing_handler.go +++ b/handler/platform_billing_handler.go @@ -1,6 +1,8 @@ package handler import ( + "net/http" + "github.com/gin-gonic/gin" "laic-backend/common" @@ -8,6 +10,25 @@ import ( "laic-backend/vo" ) +// GrantTrafficForTest adds traffic to a regular user without a payment order. +func GrantTrafficForTest(c *gin.Context) { + if common.AppConf == nil || !common.AppConf.Testing.AllowTrafficGrant { + common.FailWithBusiErrorWithHttpStatus(c, http.StatusForbidden, common.ErrForbidden) + return + } + var req vo.TrafficTestGrantReq + if err := c.ShouldBindJSON(&req); err != nil { + common.FailWithBindError(c, common.ErrParam, err) + return + } + data, e := service.DefaultBillingService.GrantTrafficForTest(common.GetUserId(c), req.UserID, req.AmountGb, req.Reason) + if e != nil { + common.FailWithBusiError(c, e) + return + } + common.OKWithData(c, data) +} + func GetPlatformResourcePool(c *gin.Context) { data, e := service.DefaultPlatformBillingService.GetPool() if e != nil { diff --git a/mock/dock.go b/mock/dock.go index 5bffccf..8f196de 100644 --- a/mock/dock.go +++ b/mock/dock.go @@ -1,9 +1,12 @@ package main import ( + "context" "encoding/json" "fmt" "log" + "os" + "os/exec" "strings" "sync" "time" @@ -50,28 +53,51 @@ type droneTelemetry struct { // MockDock 模拟机巢：通过 MQTT 上云并直接上报机巢、无人机状态 type MockDock struct { - spec dockSpec + spec dockSpec + clientIDPrefix string + videoFile string + bootID string client paho.Client - mu sync.Mutex - doorState string - droneOnline bool - tele droneTelemetry - inMission bool - missionPaused bool - missionCancel chan struct{} - missionCmdID string - uploadedRoute []missionWaypoint - liveSessionID string - liveStreaming bool - videoVersion int64 -} - -func newMockDock(spec dockSpec) *MockDock { + mu sync.Mutex + doorState string + droneOnline bool + tele droneTelemetry + inMission bool + missionPaused bool + missionCancel chan struct{} + missionCmdID string + uploadedRoute []missionWaypoint + liveSessionID string + liveStreaming bool + liveProvider string + liveProtocol string + liveMaxBitrateBps int64 + liveError string + liveStopReason string + liveCancel context.CancelFunc + liveProcess *exec.Cmd + commandResults map[string]commandResult + commandPending map[string]chan struct{} + publishStop chan struct{} + publishDone chan struct{} + stopOnce sync.Once + videoVersion int64 +} + +type commandResult struct { + accepted bool + resultCode string +} + +func newMockDock(spec dockSpec, clientIDPrefix, videoFile string) *MockDock { d := &MockDock{ - spec: spec, - doorState: "closed", + spec: spec, + clientIDPrefix: clientIDPrefix, + videoFile: videoFile, + bootID: fmt.Sprintf("boot-%s-%d", spec.DockID, time.Now().UnixNano()), + doorState: "closed", tele: droneTelemetry{ flightMode: "STANDBY", latitude: spec.DroneLat, @@ -96,25 +122,63 @@ func newMockDock(spec dockSpec) *MockDock { func (d *MockDock) start(broker, username, password string) error { // 1. MQTT 连接后台 EMQX + offline, err := json.Marshal(d.wrap("", "", map[string]any{ + "status": "offline", + "bootId": d.bootID, + "dockIdSource": "dmi_product_serial", + "softwareVersion": "1.3.0", + "protocolVersion": "1.0", + "timeSynced": true, + "mqttConnected": false, + "modbusConnected": false, + "mavlinkConnected": false, + "updating": false, + })) + if err != nil { + return err + } opts := paho.NewClientOptions(). AddBroker(broker). - SetClientID("mock-dock-" + d.spec.DockID). + SetClientID(d.clientIDPrefix + d.spec.DockID). SetUsername(username). SetPassword(password). SetAutoReconnect(true). SetCleanSession(true) + opts.SetWill(d.topic("status/online"), string(offline), 1, true) opts.SetOnConnectHandler(func(c paho.Client) { d.onMqttConnect(c) }) d.client = paho.NewClient(opts) if tok := d.client.Connect(); tok.Wait() && tok.Error() != nil { return fmt.Errorf("MQTT 连接失败: %w", tok.Error()) } + d.publishStop = make(chan struct{}) + d.publishDone = make(chan struct{}) go d.publishLoop() log.Printf("[%s] 机巢启动（MQTT 直接模拟无人机数据）", d.spec.DockID) return nil } +// stop publishes an orderly offline status, stops any media publisher, and disconnects MQTT. +func (d *MockDock) stop() { + d.stopOnce.Do(func() { + if d.publishStop != nil { + close(d.publishStop) + } + }) + d.publishStatus("offline") + d.stopLivePublisher() + if d.publishDone != nil { + select { + case <-d.publishDone: + case <-time.After(time.Second): + } + } + if d.client != nil && d.client.IsConnected() { + d.client.Disconnect(250) + } +} + func (d *MockDock) topic(sub string) string { return fmt.Sprintf("dock-edge/v1/dock/%s/%s", d.spec.DockID, sub) } @@ -146,7 +210,7 @@ func (d *MockDock) publishVideoState(payload any, version int64) { } b, err := json.Marshal(map[string]any{ "requestId": nil, - "eventId": fmt.Sprintf("video-%s-%d", d.spec.DockID, version), + "eventId": fmt.Sprintf("video-%s-%s-%d", d.spec.DockID, d.bootID, version), "version": version, "dockId": d.spec.DockID, "droneSn": nil, @@ -186,8 +250,11 @@ func (d *MockDock) publishLoop() { tick5 := time.NewTicker(5 * time.Second) defer tick1.Stop() defer tick5.Stop() + defer close(d.publishDone) for { select { + case <-d.publishStop: + return case <-tick1.C: d.publishTelemetry() case <-tick5.C: @@ -215,6 +282,9 @@ func (d *MockDock) onCommand(_ paho.Client, msg paho.Message) { cmd.DroneSN = *env.DroneSN } log.Printf("[%s] 收到指令 %s (commandId=%s)", d.spec.DockID, cmd.Type, cmd.CommandID) + if d.replayCommand(cmd) { + return + } switch { case strings.HasPrefix(cmd.Type, "dock."): @@ -226,7 +296,7 @@ func (d *MockDock) onCommand(_ paho.Client, msg paho.Message) { case strings.HasPrefix(cmd.Type, "video."): d.handleVideoCommand(cmd) default: - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, false, "UNSUPPORTED_COMMAND") + d.ackCommand(cmd, false, "UNSUPPORTED_COMMAND") } } @@ -243,11 +313,11 @@ func (d *MockDock) handleDockCommand(cmd commandMsg) { "dock.centering_loose", "dock.centering_tight", "dock.clear_alarm", "dock.emergency_stop": // 无状态变化，仅确认 default: - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, false, "UNSUPPORTED_COMMAND") + d.ackCommand(cmd, false, "UNSUPPORTED_COMMAND") return } d.publishStateDock() - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, true, "OK") + d.ackCommand(cmd, true, "OK") } func (d *MockDock) handleDroneCommand(cmd commandMsg) { @@ -263,7 +333,7 @@ func (d *MockDock) handleDroneCommand(cmd commandMsg) { case "drone.land", "drone.return": if d.cancelMission() { // 任务协程收到取消后会自行返航落地 - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, true, "OK") + d.ackCommand(cmd, true, "OK") return } d.snapHomeLanded() @@ -278,14 +348,14 @@ func (d *MockDock) handleDroneCommand(cmd commandMsg) { wps := append([]missionWaypoint(nil), d.uploadedRoute...) d.mu.Unlock() if len(wps) == 0 { - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, false, "NO_ROUTE") + d.ackCommand(cmd, false, "NO_ROUTE") return } if !d.beginMission(cmd.CommandID) { - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, false, "MISSION_IN_PROGRESS") + d.ackCommand(cmd, false, "MISSION_IN_PROGRESS") return } - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, true, "OK") + d.ackCommand(cmd, true, "OK") go d.flyUploadedMission(wps) return case "drone.mission_pause": @@ -295,12 +365,12 @@ func (d *MockDock) handleDroneCommand(cmd commandMsg) { case "drone.mission_cancel": d.cancelMission() default: - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, false, "UNSUPPORTED_COMMAND") + d.ackCommand(cmd, false, "UNSUPPORTED_COMMAND") return } d.publishStateDrone() d.publishStateDock() - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, true, "OK") + d.ackCommand(cmd, true, "OK") } func (d *MockDock) handleVideoCommand(cmd commandMsg) { @@ -308,35 +378,179 @@ func (d *MockDock) handleVideoCommand(cmd commandMsg) { switch cmd.Type { case "video.start_stream": if streamSessionID == "" { - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, false, "SESSION_ID_REQUIRED") + d.ackCommand(cmd, false, "SESSION_ID_REQUIRED") return } + pushURL, _ := cmd.Params["pushUrl"].(string) + provider, _ := cmd.Params["provider"].(string) + maxBitrate := asInt64(cmd.Params["maxBitrateBps"]) + if maxBitrate <= 0 { + maxBitrate = 1500000 + } d.mu.Lock() if d.liveStreaming && d.liveSessionID != streamSessionID { d.mu.Unlock() - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, false, "STREAM_IN_PROGRESS") + d.ackCommand(cmd, false, "STREAM_IN_PROGRESS") return } d.liveSessionID = streamSessionID d.liveStreaming = true + d.liveProvider = provider + d.liveProtocol = pushProtocol(pushURL) + d.liveMaxBitrateBps = maxBitrate + d.liveStopReason = "" + d.liveError = "" d.mu.Unlock() - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, true, "OK") + if err := d.startLivePublisher(pushURL); err != nil { + d.mu.Lock() + d.liveStreaming = false + d.liveError = "PUBLISHER_START_FAILED" + d.mu.Unlock() + d.ackCommand(cmd, false, "PUBLISHER_START_FAILED") + d.publishStateVideo() + return + } + d.ackCommand(cmd, true, "OK") d.publishStateVideo() case "video.stop_stream": d.mu.Lock() matches := streamSessionID != "" && streamSessionID == d.liveSessionID if matches { d.liveStreaming = false + d.liveStopReason, _ = cmd.Params["reason"].(string) + if d.liveStopReason == "" { + d.liveStopReason = "device_request" + } } d.mu.Unlock() if !matches { - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, false, "SESSION_NOT_FOUND") + d.ackCommand(cmd, false, "SESSION_NOT_FOUND") return } - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, true, "OK") + d.stopLivePublisher() + d.ackCommand(cmd, true, "OK") d.publishStateVideo() default: - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, false, "UNSUPPORTED_COMMAND") + d.ackCommand(cmd, false, "UNSUPPORTED_COMMAND") + } +} + +// startLivePublisher starts ffmpeg for real SRT/RTMP(S) push URLs. Fake URLs +// intentionally keep the control-plane simulation without requiring ffmpeg. +func (d *MockDock) startLivePublisher(pushURL string) error { + if strings.HasPrefix(strings.ToLower(pushURL), "fake://") || pushURL == "" { + return nil + } + if !isSupportedPushURL(pushURL) { + return fmt.Errorf("unsupported push URL scheme: %s", pushURL) + } + if d.videoFile == "" { + return fmt.Errorf("video file is not configured") + } + if _, err := os.Stat(d.videoFile); err != nil { + return fmt.Errorf("video file unavailable: %w", err) + } + ctx, cancel := context.WithCancel(context.Background()) + ffmpeg := envOr("MOCK_FFMPEG_BIN", "ffmpeg") + d.mu.Lock() + maxBitrate := d.liveMaxBitrateBps + if d.liveProcess != nil { + d.mu.Unlock() + cancel() + return nil + } + d.mu.Unlock() + cmd := exec.CommandContext(ctx, ffmpeg, ffmpegArgs(d.videoFile, pushURL, maxBitrate)...) + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + d.mu.Lock() + if d.liveProcess != nil { + d.mu.Unlock() + cancel() + return nil + } + d.liveCancel = cancel + d.liveProcess = cmd + d.mu.Unlock() + if err := cmd.Start(); err != nil { + cancel() + d.mu.Lock() + d.liveCancel = nil + d.liveProcess = nil + d.mu.Unlock() + return err + } + go d.waitLivePublisher(cmd) + return nil +} + +func (d *MockDock) waitLivePublisher(cmd *exec.Cmd) { + err := cmd.Wait() + d.mu.Lock() + if d.liveProcess != cmd { + d.mu.Unlock() + return + } + d.liveProcess = nil + d.liveCancel = nil + wasStreaming := d.liveStreaming + d.liveStreaming = false + if err != nil { + d.liveError = "PUBLISHER_EXITED" + } + d.mu.Unlock() + if wasStreaming { + d.publishStateVideo() + } +} + +func (d *MockDock) stopLivePublisher() { + d.mu.Lock() + cancel := d.liveCancel + d.liveCancel = nil + d.liveProcess = nil + d.mu.Unlock() + if cancel != nil { + cancel() + } +} + +func isSupportedPushURL(pushURL string) bool { + url := strings.ToLower(pushURL) + return strings.HasPrefix(url, "srt://") || strings.HasPrefix(url, "rtmp://") || strings.HasPrefix(url, "rtmps://") +} + +func ffmpegArgs(videoFile, pushURL string, maxBitrateBps int64) []string { + if maxBitrateBps <= 0 { + maxBitrateBps = 1500000 + } + bitrate := fmt.Sprintf("%d", maxBitrateBps) + bufsize := fmt.Sprintf("%d", maxBitrateBps*2) + args := []string{"-hide_banner", "-loglevel", "warning", "-re", "-stream_loop", "-1", "-i", videoFile, + "-map", "0:v:0", "-map", "0:a:0?", + "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", "-pix_fmt", "yuv420p", + "-b:v", bitrate, "-maxrate", bitrate, "-bufsize", bufsize, "-g", "60", + "-c:a", "aac", "-b:a", "128k", + } + if strings.HasPrefix(strings.ToLower(pushURL), "srt://") { + return append(args, "-f", "mpegts", pushURL) + } + return append(args, "-f", "flv", pushURL) +} + +func pushProtocol(pushURL string) string { + url := strings.ToLower(pushURL) + switch { + case strings.HasPrefix(url, "srt://"): + return "srt" + case strings.HasPrefix(url, "rtmp://"): + return "rtmp" + case strings.HasPrefix(url, "rtmps://"): + return "rtmps" + case strings.HasPrefix(url, "fake://"): + return "fake" + default: + return "" } } @@ -344,25 +558,25 @@ func (d *MockDock) handleWorkflowCommand(cmd commandMsg) { switch cmd.Type { case "workflow.cancel", "workflow.stop_task": d.cancelMission() - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, true, "OK") + d.ackCommand(cmd, true, "OK") return case "workflow.one_key_return", "workflow.one_key_landing": if !d.cancelMission() { go d.rtlAndLand() } - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, true, "OK") + d.ackCommand(cmd, true, "OK") return } if d.isRepeatCommand(cmd.CommandID) { - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, true, "OK") + d.ackCommand(cmd, true, "OK") return } if !d.beginMission(cmd.CommandID) { - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, false, "MISSION_IN_PROGRESS") + d.ackCommand(cmd, false, "MISSION_IN_PROGRESS") return } - d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, true, "OK") + d.ackCommand(cmd, true, "OK") switch cmd.Type { case "workflow.start_task": @@ -390,17 +604,67 @@ func (d *MockDock) ack(requestID, droneSN, commandID string, accepted bool, resu }) } +// replayCommand implements the commandId idempotency expected from an edge device. +// Retries carry a new requestId, so the original result is acknowledged with the +// current requestId without executing the command a second time. Concurrent +// duplicates wait for the first execution to publish its result. +func (d *MockDock) replayCommand(cmd commandMsg) bool { + if cmd.CommandID == "" { + return false + } + d.mu.Lock() + result, ok := d.commandResults[cmd.CommandID] + if !ok { + if pending, exists := d.commandPending[cmd.CommandID]; exists { + d.mu.Unlock() + <-pending + return d.replayCommand(cmd) + } + if d.commandPending == nil { + d.commandPending = make(map[string]chan struct{}) + } + d.commandPending[cmd.CommandID] = make(chan struct{}) + } + d.mu.Unlock() + if !ok { + return false + } + d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, result.accepted, result.resultCode) + return true +} + +func (d *MockDock) ackCommand(cmd commandMsg, accepted bool, resultCode string) { + if cmd.CommandID != "" { + d.mu.Lock() + if d.commandResults == nil { + d.commandResults = make(map[string]commandResult) + } + d.commandResults[cmd.CommandID] = commandResult{accepted: accepted, resultCode: resultCode} + if pending, ok := d.commandPending[cmd.CommandID]; ok { + delete(d.commandPending, cmd.CommandID) + close(pending) + } + d.mu.Unlock() + } + d.ack(cmd.RequestID, cmd.DroneSN, cmd.CommandID, accepted, resultCode) +} + func (d *MockDock) publishStatusOnline() { + d.publishStatus("online") +} + +func (d *MockDock) publishStatus(status string) { + online := status == "online" d.publish(d.topic("status/online"), 1, true, map[string]any{ - "status": "online", - "bootId": "boot-" + d.spec.DockID, + "status": status, + "bootId": d.bootID, "dockIdSource": "dmi_product_serial", "softwareVersion": "1.3.0", "protocolVersion": "1.0", "uptimeSec": 86400, "timeSynced": true, - "mqttConnected": true, - "modbusConnected": true, + "mqttConnected": online, + "modbusConnected": online, "mavlinkConnected": false, "updating": false, "name": d.spec.Name, @@ -542,6 +806,11 @@ func (d *MockDock) publishStateVideo() { d.mu.Lock() streamSessionID := d.liveSessionID streaming := d.liveStreaming + provider := d.liveProvider + protocol := d.liveProtocol + maxBitrate := d.liveMaxBitrateBps + errorCode := d.liveError + stopReason := d.liveStopReason d.videoVersion++ version := d.videoVersion d.mu.Unlock() @@ -550,28 +819,36 @@ func (d *MockDock) publishStateVideo() { if streaming { phase = "streaming" } - provider := "fake" + if provider == "" { + provider = "fake" + } + if protocol == "" { + protocol = "srt" + } d.publishVideoState(map[string]any{ "provider": provider, "phase": phase, "inputOnline": streaming, "streaming": streaming, "inputCodec": "h264", - "uplinkProtocol": "srt", + "uplinkProtocol": protocol, "streamSessionId": nilIfEmpty(streamSessionID), "width": nil, "height": nil, "frameRate": nil, - "bitrateBps": liveBitrate(streaming), + "bitrateBps": liveBitrate(streaming, maxBitrate), "retryCount": 0, - "stopReason": nil, - "errorCode": nil, + "stopReason": nilIfEmpty(stopReason), + "errorCode": nilIfEmpty(errorCode), "updatedAt": time.Now().UnixMilli(), }, version) } -func liveBitrate(streaming bool) int { +func liveBitrate(streaming bool, maxBitrateBps int64) int64 { if streaming { + if maxBitrateBps > 0 { + return maxBitrateBps + } return 1500000 } return 0 diff --git a/mock/main.go b/mock/main.go index 3cc3287..9543fa4 100644 --- a/mock/main.go +++ b/mock/main.go @@ -4,6 +4,7 @@ import ( "log" "os" "os/signal" + "path/filepath" "strings" "syscall" "time" @@ -26,9 +27,16 @@ type dockSpec struct { } func main() { - broker := envOr("MOCK_MQTT_BROKER", "tcp://127.0.0.1:1883") + broker := envOr("MOCK_MQTT_BROKER", "tcp://roll.jiagutech.com:1883") username := envOr("MOCK_MQTT_USERNAME", "laic") password := envOr("MOCK_MQTT_PASSWORD", "") + clientIDPrefix := envOr("MOCK_MQTT_CLIENT_ID_PREFIX", "mock-dock-") + videoFile := resolveVideoFile(envOr("MOCK_VIDEO_FILE", "mock/mock.mp4")) + if info, err := os.Stat(videoFile); err != nil { + log.Printf("视频素材不可用（仅影响真实推流和原始视频上传）: %s: %v", videoFile, err) + } else { + log.Printf("视频素材: %s (%d bytes)", videoFile, info.Size()) + } specs := []dockSpec{ { @@ -51,11 +59,13 @@ func main() { }, } + docks := make([]*MockDock, 0, len(specs)) for _, spec := range selectDockSpecs(specs, os.Getenv("MOCK_DOCKS")) { - dock := newMockDock(spec) + dock := newMockDock(spec, clientIDPrefix, videoFile) if err := dock.start(broker, username, password); err != nil { log.Fatalf("启动机巢 %s 失败: %v", spec.DockID, err) } + docks = append(docks, dock) } log.Println("模拟设备已启动（机巢 MQTT 直接上报机巢与无人机数据），按 Ctrl+C 退出") @@ -63,9 +73,30 @@ func main() { signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) <-quit log.Println("正在退出...") + for _, dock := range docks { + dock.stop() + } time.Sleep(200 * time.Millisecond) } +// resolveVideoFile first checks the current working directory, then the executable directory. +// This keeps the default path usable both with `go run` and with a copied binary. +func resolveVideoFile(path string) string { + if filepath.IsAbs(path) { + return path + } + if _, err := os.Stat(path); err == nil { + return path + } + if exe, err := os.Executable(); err == nil { + candidate := filepath.Join(filepath.Dir(exe), path) + if _, err := os.Stat(candidate); err == nil { + return candidate + } + } + return path +} + func selectDockSpecs(specs []dockSpec, selected string) []dockSpec { if strings.TrimSpace(selected) == "" { return specs diff --git a/mock/mission.go b/mock/mission.go index cd4ad6e..8123880 100644 --- a/mock/mission.go +++ b/mock/mission.go @@ -1,11 +1,11 @@ package main import ( - "bytes" "io" "log" "math" "net/http" + "os" "sort" "strconv" "time" @@ -382,12 +382,23 @@ func (d *MockDock) uploadOriginalVideo(cmd commandMsg, taskID string, video orig return } - body := []byte("mock original video fixture\n") - req, err := http.NewRequest(http.MethodPut, video.UploadURL, bytes.NewReader(body)) + file, err := os.Open(d.videoFile) + if err != nil { + d.publishOriginalVideo(cmd, taskID, video, "failed", "VIDEO_FILE_NOT_FOUND", 0) + return + } + defer file.Close() + info, err := file.Stat() + if err != nil || info.Size() <= 0 { + d.publishOriginalVideo(cmd, taskID, video, "failed", "VIDEO_FILE_INVALID", 0) + return + } + req, err := http.NewRequest(http.MethodPut, video.UploadURL, file) if err != nil { d.publishOriginalVideo(cmd, taskID, video, "failed", "OSS_PUT_FAILED", 0) return } + req.ContentLength = info.Size() req.Header.Set("Content-Type", "video/mp4") resp, err := (&http.Client{Timeout: 20 * time.Second}).Do(req) if err != nil { @@ -400,7 +411,7 @@ func (d *MockDock) uploadOriginalVideo(cmd commandMsg, taskID string, video orig d.publishOriginalVideo(cmd, taskID, video, "failed", "OSS_PUT_FAILED", 0) return } - d.publishOriginalVideo(cmd, taskID, video, "completed", "", int64(len(body))) + d.publishOriginalVideo(cmd, taskID, video, "completed", "", info.Size()) } func (d *MockDock) publishOriginalVideo(cmd commandMsg, taskID string, video originalVideoUpload, eventType, errorCode string, fileSize int64) { diff --git a/model/billing.go b/model/billing.go index d7c624f..35f13bd 100644 --- a/model/billing.go +++ b/model/billing.go @@ -20,7 +20,7 @@ type TrafficOrder struct { CreditLedgerID int64 `gorm:"column:credit_ledger_id;type:BIGINT" json:"creditLedgerId"` CreditAttemptCount int `gorm:"column:credit_attempt_count;type:INT;not null" json:"-"` NextCreditAt *time.Time `gorm:"column:next_credit_at" json:"-"` - PaymentIdempotencyKey string `gorm:"column:payment_idempotency_key;type:VARCHAR(128);uniqueIndex:uk_traffic_order_payment_idem" json:"-"` + PaymentIdempotencyKey *string `gorm:"column:payment_idempotency_key;type:VARCHAR(128);uniqueIndex:uk_traffic_order_payment_idem" json:"-"` PaidBy int64 `gorm:"column:paid_by;type:BIGINT" json:"paidBy"` CreatedAt time.Time `gorm:"column:created_at" json:"createdAt"` PaidAt *time.Time `gorm:"column:paid_at" json:"paidAt"` diff --git a/route/route.go b/route/route.go index eaf2b2f..21df179 100644 --- a/route/route.go +++ b/route/route.go @@ -189,6 +189,7 @@ func InitRouter(port int32) { { trafficAdmin.GET("/pool", handler.GetPlatformResourcePool) trafficAdmin.POST("/purchases", handler.CreatePlatformPurchase) + trafficAdmin.POST("/test-grants", handler.GrantTrafficForTest) trafficAdmin.GET("/purchases", handler.GetPlatformPurchasePage) trafficAdmin.GET("/packages", handler.GetTrafficPackagePage) trafficAdmin.POST("/packages", handler.CreateTrafficPackage) diff --git a/service/billing_service.go b/service/billing_service.go index edb3159..f77d52b 100644 --- a/service/billing_service.go +++ b/service/billing_service.go @@ -24,6 +24,67 @@ const ( defaultTrafficUnitPrice = 10.0 // 云媒体流量单价（元/GB） ) +// GrantTrafficForTest credits a regular user's balance through the same +// snapshot and ledger fields used by paid traffic fulfillment. It is guarded +// by an explicit testing configuration switch and is intended for QA only. +func (b *BillingService) GrantTrafficForTest(operatorID, userID, amountGb int64, reason string) (*vo.TrafficBalanceVO, *common.BusiError) { + if common.AppConf == nil || !common.AppConf.Testing.AllowTrafficGrant { + return nil, common.ErrForbidden + } + if operatorID <= 0 || userID <= 0 || amountGb <= 0 || amountGb > int64(^uint64(0)>>1)/gbBytes { + return nil, common.ErrParam + } + + amountBytes := amountGb * gbBytes + now := time.Now() + var after int64 + err := common.DB.Transaction(func(tx *gorm.DB) error { + var user model.User + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&user, userID).Error; err != nil { + return err + } + if user.Role != common.RoleUser { + return common.NewBusiError(common.ParamError, "仅支持给普通用户添加测试流量") + } + if user.TrafficBalance > int64(^uint64(0)>>1)-amountBytes { + return common.NewBusiError(common.ParamError, "流量余额超出系统上限") + } + after = user.TrafficBalance + amountBytes + if err := tx.Model(&user).UpdateColumn("traffic_balance", after).Error; err != nil { + return err + } + ledgerID := mustID() + if ledgerID == 0 { + return errors.New("generate traffic grant ledger ID") + } + if reason == "" { + reason = "test grant" + } + return tx.Create(&model.TrafficLedger{ + ID: ledgerID, AccountType: "user", AccountID: userID, Direction: "credit", + AmountBytes: amountBytes, BalanceBefore: user.TrafficBalance, BalanceAfter: after, + SourceType: "test_grant", SourceID: fmt.Sprintf("admin:%d", operatorID), + IdempotencyKey: fmt.Sprintf("test-grant:%d", ledgerID), OperatorID: operatorID, + Remark: reason, CreatedAt: now, + }).Error + }) + if err != nil { + if busiErr, ok := err.(*common.BusiError); ok { + return nil, busiErr + } + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, common.ErrUserNotFound + } + logger.ERROR("测试流量入账事务失败", err) + return nil, common.ErrInternal + } + if err := common.Delete(cache.TrafficKeyOf(userID)); err != nil { + logger.WARN("删除测试流量余额缓存失败", err) + } + DefaultOperationLogService.RecordEvent(operatorID, "充值与履约", "测试流量入账", fmt.Sprintf("targetUserId=%d amountGb=%d", userID, amountGb), "success", "admin_test") + return &vo.TrafficBalanceVO{BalanceBytes: after, BalanceGb: float64(after) / float64(gbBytes)}, nil +} + // trafficDeductLua 原子扣减：key 不存在时用 MySQL 快照兜底初始化，再判断并扣减。 // 返回 {status, before, after}：status=1 成功，0 余额不足。 const trafficDeductLua = ` @@ -261,6 +322,7 @@ func (b *BillingService) CreateOrderWithPackage(userID, packageID int64, amountG } var pending model.TrafficOrder if err := tx.Where("user_id = ? AND pay_status IN ?", userID, []string{"unpaid", "processing"}).First(&pending).Error; err == nil { + logger.WARN("创建流量订单被待支付订单阻止", "userId", userID, "pendingOrderId", pending.ID, "payStatus", pending.PayStatus) busiErr = common.ErrPendingOrderExists return busiErr } else if !errors.Is(err, gorm.ErrRecordNotFound) { @@ -293,7 +355,7 @@ func (b *BillingService) CreateOrderWithPackage(userID, packageID int64, amountG if errors.Is(err, gorm.ErrRecordNotFound) { return nil, common.ErrUserNotFound } - if code, _ := common.ParseError(err); code == 1062 { + if code, constraint := common.ParseError(err); code == 1062 && constraint == "uk_traffic_order_pending_user" { return nil, common.ErrPendingOrderExists } logger.ERROR("创建流量订单失败", err) @@ -307,6 +369,12 @@ func (b *BillingService) CancelOrder(userID, orderID int64) *common.BusiError { now := time.Now() var busiErr *common.BusiError err := common.DB.Transaction(func(tx *gorm.DB) error { + // Serialize cancellation with CreateOrderWithPackage, which takes the + // same user-row lock before checking for pending orders. + var user model.User + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id").First(&user, userID).Error; err != nil { + return err + } var order model.TrafficOrder if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ? AND user_id = ?", orderID, userID).First(&order).Error; err != nil { return err @@ -465,10 +533,12 @@ func (b *BillingService) chargeLiveSession(session *model.LiveSession, now time. periodStart := *current.StartedAt var last model.LiveBillingSegment - if err := tx.Where("session_id = ?", current.ID).Order("period_end DESC").First(&last).Error; err == nil { + // A missing prior segment is expected for the first billing tick. Use + // Find so GORM does not emit a misleading record-not-found error. + if result := tx.Where("session_id = ?", current.ID).Order("period_end DESC").Limit(1).Find(&last); result.Error != nil { + return result.Error + } else if result.RowsAffected == 1 { periodStart = last.PeriodEnd - } else if !errors.Is(err, gorm.ErrRecordNotFound) { - return err } if !now.After(periodStart) { return nil diff --git a/service/live_service.go b/service/live_service.go index a7c8dbf..be2dcea 100644 --- a/service/live_service.go +++ b/service/live_service.go @@ -9,6 +9,7 @@ import ( "time" "gorm.io/gorm" + "gorm.io/gorm/clause" "laic-backend/cache" "laic-backend/common" @@ -70,8 +71,11 @@ func (s *LiveService) Join(userID int64, isAdmin bool, dockID string, req *vo.Li if err := tx.Set("gorm:query_option", "FOR UPDATE").Where("dock_id = ?", dockID).First(&model.Dock{}).Error; err != nil { return err } - if err := tx.Where("dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", dockID).Order("created_at DESC, id DESC").First(&session).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { - return err + // 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). + Order("created_at DESC, id DESC").Limit(1).Find(&session); result.Error != nil { + return result.Error } if session.ID == "" { maxBitrate := req.MaxBitrateBps @@ -105,7 +109,14 @@ func (s *LiveService) Join(userID int64, isAdmin bool, dockID string, req *vo.Li return err } lease = model.LiveViewerLease{ID: leaseID, StreamSessionID: session.ID, ViewerID: userID, ExpiresAt: now.Add(time.Duration(leaseSeconds()) * time.Second), CreatedAt: now, UpdatedAt: now} - return tx.Where("stream_session_id = ? AND viewer_id = ?", session.ID, userID).Assign(map[string]any{"expires_at": lease.ExpiresAt, "released_at": nil, "updated_at": now}).FirstOrCreate(&lease).Error + return tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "stream_session_id"}, {Name: "viewer_id"}}, + DoUpdates: clause.Assignments(map[string]any{ + "expires_at": lease.ExpiresAt, + "released_at": nil, + "updated_at": now, + }), + }).Create(&lease).Error }) if err != nil { logger.ERROR("创建直播观看租约失败", err) @@ -387,7 +398,19 @@ func (s *LiveService) GetPlayURL(userID int64, isAdmin bool, dockID, sessionID s return nil, busiErr } var session model.LiveSession - if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("id = ? AND dock_id = ? AND phase = 'streaming'", sessionID, dockID).First(&session).Error; err != nil { + query := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("dock_id = ? AND phase = 'streaming'", dockID) + if sessionID != "" { + query = query.Where("id = ?", sessionID) + } else { + // Keep compatibility with older clients that did not send streamSessionId. + // The lease predicate prevents selecting another user's live session. + query = query.Where(`id IN ( + SELECT stream_session_id + FROM live_viewer_lease + WHERE viewer_id = ? AND released_at IS NULL AND expires_at > ? + )`, userID, time.Now()).Order("created_at DESC, id DESC") + } + if err := query.First(&session).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, common.ErrLiveNotFound } @@ -457,6 +480,16 @@ func (s *LiveService) OnVideoState(dockID string, streaming bool, streamSessionI if version <= session.DeviceStateVersion || (session.DeviceEventID != "" && session.DeviceEventID == eventID) || updatedAt < session.DeviceUpdatedAt { return false } + // eventId is expected to be globally unique. Reject a stale/reused device + // event that was already recorded for another live session before touching + // the unique device-event column. + var eventOwner model.LiveSession + if err := common.DB.Select("id").Where("device_event_id = ? AND id <> ?", eventID, streamSessionID).First(&eventOwner).Error; err == nil { + logger.WARN("忽略已被其他直播会话使用的视频状态事件", eventID, streamSessionID, eventOwner.ID) + return false + } else if !errors.Is(err, gorm.ErrRecordNotFound) { + return false + } updates := map[string]any{ "device_state_version": version, "device_event_id": eventID, @@ -472,7 +505,14 @@ func (s *LiveService) OnVideoState(dockID string, streaming bool, streamSessionI online, queryErr := provider.QueryOnline(session.StreamName) if queryErr != nil || !online.Online { updates["error_code"] = "PROVIDER_ONLINE_UNVERIFIED" - return common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND device_state_version < ?", streamSessionID, dockID, version).Updates(updates).Error == nil + result := common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND device_state_version < ?", streamSessionID, dockID, version).Updates(updates) + if result.Error != nil { + if code, constraint := common.ParseError(result.Error); code == 1062 && constraint == "uk_live_session_device_event" { + return false + } + return false + } + return true } updates["phase"] = "streaming" updates["started_at"] = now @@ -484,7 +524,13 @@ func (s *LiveService) OnVideoState(dockID string, streaming bool, streamSessionI updates["error_code"] = "DEVICE_STREAM_OFFLINE" } result := common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping') AND device_state_version < ?", streamSessionID, dockID, version).Updates(updates) - return result.Error == nil && result.RowsAffected == 1 + if result.Error != nil { + if code, constraint := common.ParseError(result.Error); code == 1062 && constraint == "uk_live_session_device_event" { + return false + } + return false + } + return result.RowsAffected == 1 } func abs(value int64) int64 { diff --git a/service/payment_service.go b/service/payment_service.go index 67751e3..e4f557c 100644 --- a/service/payment_service.go +++ b/service/payment_service.go @@ -2,10 +2,14 @@ package service import ( "crypto/sha256" + "encoding/base64" "encoding/hex" + "encoding/json" "errors" "fmt" + "net/http" "strconv" + "strings" "time" "gorm.io/gorm" @@ -66,12 +70,16 @@ func (s *PaymentService) CreateWechatPayment(userID int64, req *vo.PaymentCreate return nil, common.ErrPaymentAlreadyPaid } if !created { - if transaction.Status == "cancelled" || transaction.Status == "closed" || transaction.Status == "failed" || transaction.ProviderPayload == "" { + if transaction.Status == "cancelled" || transaction.Status == "closed" || transaction.Status == "failed" { DefaultOperationLogService.RecordEvent(userID, "充值与履约", "复用支付交易", fmt.Sprintf("transactionId=%d businessType=%s businessOrderId=%d status=%s", transaction.ID, transaction.BusinessType, transaction.BusinessOrderID, transaction.Status), "failed", "user_api") return nil, common.ErrPaymentOrderConflict } - DefaultOperationLogService.RecordEvent(userID, "充值与履约", "复用支付交易", fmt.Sprintf("transactionId=%d businessType=%s businessOrderId=%d status=%s", transaction.ID, transaction.BusinessType, transaction.BusinessOrderID, transaction.Status), "success", "user_api") - return paymentCreateVO(transaction, transaction.ProviderPayload), nil + if isUsableProviderPayload(transaction.ProviderPayload) { + DefaultOperationLogService.RecordEvent(userID, "充值与履约", "复用支付交易", fmt.Sprintf("transactionId=%d businessType=%s businessOrderId=%d status=%s", transaction.ID, transaction.BusinessType, transaction.BusinessOrderID, transaction.Status), "success", "user_api") + return paymentCreateVO(transaction, transaction.ProviderPayload), nil + } + // Older versions stored raw PNG bytes as UTF-8. The stale payload is + // replaced below by requesting a fresh QR code for this transaction. } payload, err := s.agpay.CreateWechatQRCode(client.AGPayCreateRequest{ @@ -81,7 +89,11 @@ func (s *PaymentService) CreateWechatPayment(userID int64, req *vo.PaymentCreate logger.WARN("创建 AGPay 微信支付请求失败，交易保留待确认", transaction.ID, err) return nil, common.ErrPaymentOrderConflict } - payloadText := string(payload) + payloadText := encodeProviderPayload(payload) + if !isUsableProviderPayload(payloadText) { + logger.WARN("AGPay 返回了无法展示的支付二维码", transaction.ID) + return nil, common.ErrPaymentOrderConflict + } if err := common.DB.Model(&model.PaymentTransaction{}).Where("id = ? AND status = ?", transaction.ID, "processing").Updates(map[string]any{ "provider_payload": payloadText, "requested_at": time.Now(), "updated_at": time.Now(), }).Error; err != nil { @@ -105,6 +117,37 @@ func (s *PaymentService) GetTransaction(userID, transactionID int64) (*vo.Paymen return paymentTransactionVO(&transaction), nil } +// encodeProviderPayload converts AGPay's binary QR image into a JSON-safe data +// URI. JSON responses remain unchanged for compatibility with other gateways. +func encodeProviderPayload(payload []byte) string { + if len(payload) == 0 { + return "" + } + if json.Valid(payload) { + return string(payload) + } + contentType := http.DetectContentType(payload) + return "data:" + contentType + ";base64," + base64.StdEncoding.EncodeToString(payload) +} + +func isUsableProviderPayload(payload string) bool { + if payload == "" || strings.ContainsRune(payload, '\ufffd') { + return false + } + if json.Valid([]byte(payload)) { + return true + } + if !strings.HasPrefix(payload, "data:image/") { + return false + } + comma := strings.IndexByte(payload, ',') + if comma < 0 || !strings.HasSuffix(payload[:comma], ";base64") { + return false + } + decoded, err := base64.StdEncoding.DecodeString(payload[comma+1:]) + return err == nil && len(decoded) > 0 && strings.HasPrefix(http.DetectContentType(decoded), "image/") +} + func (s *PaymentService) markPaymentRequestFailed(transactionID int64, businessType string, businessOrderID int64) { if err := common.DB.Transaction(func(tx *gorm.DB) error { if err := tx.Model(&model.PaymentTransaction{}).Where("id = ? AND status = ?", transactionID, "processing").Updates(map[string]any{"status": "failed", "updated_at": time.Now()}).Error; err != nil { diff --git a/sql/001_schema.sql b/sql/001_schema.sql index 4bf3b09..287639a 100644 --- a/sql/001_schema.sql +++ b/sql/001_schema.sql @@ -309,6 +309,7 @@ CREATE TABLE IF NOT EXISTS traffic_order ( paid_by BIGINT COMMENT '确认支付的操作人 ID', created_at DATETIME COMMENT '创建时间', paid_at DATETIME COMMENT '支付时间', + updated_at DATETIME COMMENT '更新时间', UNIQUE KEY uk_traffic_order_payment_idem (payment_idempotency_key) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/vo/platform_billing_vo.go b/vo/platform_billing_vo.go index d646923..a78fe66 100644 --- a/vo/platform_billing_vo.go +++ b/vo/platform_billing_vo.go @@ -16,6 +16,14 @@ type PlatformPurchaseCreateReq struct { Remark string `json:"remark" validate:"max=256"` } +// TrafficTestGrantReq is an admin-only test helper for crediting a regular +// user's cloud-media balance without creating a payment order. +type TrafficTestGrantReq struct { + UserID int64 `json:"userId" validate:"required,gt=0"` + AmountGb int64 `json:"amountGb" validate:"required,gt=0"` + Reason string `json:"reason" validate:"max=256"` +} + type PlatformPurchasePageReq struct { common.Pagination Status string `json:"status" form:"status"`