feat：功能增加

live/providers.go

@@ -75,10 +75,23 @@ func (a AliyunAdapter) CreateStream(req StreamRequest) (StreamCredentials, error
 func (a AliyunAdapter) CreatePlayURLs(req PlayRequest) (PlayURLs, error) {
 	expires := req.ExpiresAt
 	path := "/" + a.cfg.AppName + "/" + req.StreamName
-	signature := sign(path, expires.Unix(), a.cfg.PlayAuthKey)
-	base := "https://" + a.cfg.PlayDomain + "/" + a.cfg.AppName + "/" + req.StreamName
-	rtmp := "rtmp://" + a.cfg.PlayDomain + "/" + a.cfg.AppName + "/" + req.StreamName + "?auth_key=" + signature
-	return PlayURLs{HLS: base + ".m3u8?auth_key=" + signature, FLV: base + ".flv?auth_key=" + signature, RTMP: rtmp, ExpiresAt: expires}, nil
+	// Alibaba Live authenticates the complete playback path. HLS and FLV
+	// therefore need independent signatures because their file extensions are
+	// part of the URI used by the CDN auth check.
+	hlsPath := path + ".m3u8"
+	flvPath := path + ".flv"
+	hlsSignature := sign(hlsPath, expires.Unix(), a.cfg.PlayAuthKey)
+	flvSignature := sign(flvPath, expires.Unix(), a.cfg.PlayAuthKey)
+	base := "http://" + a.cfg.PlayDomain + "/" + a.cfg.AppName + "/" + req.StreamName
+	rtmpPath := path
+	rtmpSignature := sign(rtmpPath, expires.Unix(), a.cfg.PlayAuthKey)
+	rtmp := "rtmp://" + a.cfg.PlayDomain + "/" + a.cfg.AppName + "/" + req.StreamName + "?auth_key=" + rtmpSignature
+	return PlayURLs{
+		HLS:       base + ".m3u8?auth_key=" + hlsSignature,
+		FLV:       base + ".flv?auth_key=" + flvSignature,
+		RTMP:      rtmp,
+		ExpiresAt: expires,
+	}, nil
 }
 func (a AliyunAdapter) QueryOnline(streamName string) (OnlineStatus, error) {
 	return a.client.queryOnline(streamName)

service/billing_service.go

@@ -571,13 +571,10 @@ func (b *BillingService) chargeLiveSession(session *model.LiveSession, now time.
 			if err := tx.Model(&segment).Update("status", "insufficient").Error; err != nil {
 				return err
 			}
-			result := tx.Model(&model.LiveSession{}).Where("id = ? AND phase = 'streaming'", current.ID).Updates(map[string]any{
-				"phase": "stopping", "stop_reason": "no_balance", "error_code": "TRAFFIC_NOT_ENOUGH", "updated_at": now,
-			})
-			if result.Error != nil {
-				return result.Error
-			}
-			needStop = result.RowsAffected == 1
+			// stopForBilling performs the conditional streaming -> stopping
+			// transition after this transaction commits, and acts as the single
+			// publish gate for concurrent workers.
+			needStop = true
 			return nil
 		}
 
@@ -613,7 +610,7 @@ func (b *BillingService) chargeLiveSession(session *model.LiveSession, now time.
 		return
 	}
 	if needStop {
-		if _, busiErr := DefaultLiveService.stopForBilling(session.DockID, session.ID); busiErr != nil {
+		if _, busiErr := DefaultLiveService.stopForBilling(session.DockID, session.ID, "no_balance"); busiErr != nil {
 			logger.ERROR("直播余额不足停止推流失败", busiErr)
 		}
 		return

service/live_service.go

@@ -152,16 +152,22 @@ func (s *LiveService) GetSession(userID int64, isAdmin bool, dockID, streamSessi
 }
 func (s *LiveService) Heartbeat(userID int64, isAdmin bool, dockID, streamSessionID string) (*vo.LiveSessionVO, *common.BusiError) {
 	var session model.LiveSession
-	if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("id = ? AND dock_id = ?", streamSessionID, dockID).First(&session).Error; err != nil {
+	if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting') AND (expires_at = 0 OR expires_at > ?)", streamSessionID, dockID, time.Now().Unix()).First(&session).Error; err != nil {
 		if errors.Is(err, gorm.ErrRecordNotFound) {
 			return nil, common.ErrLiveNotFound
 		}
 		return nil, common.ErrInternal
 	}
 	expiresAt := time.Now().Add(time.Duration(leaseSeconds()) * time.Second)
-	result := common.DB.Model(&model.LiveViewerLease{}).Where("stream_session_id = ? AND viewer_id = ? AND released_at IS NULL", streamSessionID, userID).Updates(map[string]any{"expires_at": expiresAt, "updated_at": time.Now()})
+	// A heartbeat may renew only an active lease. In particular, do not
+	// resurrect a lease that the reconciler has already released or that has
+	// passed its expiry deadline.
+	now := time.Now()
+	result := common.DB.Model(&model.LiveViewerLease{}).
+		Where("stream_session_id = ? AND viewer_id = ? AND released_at IS NULL AND expires_at > ?", streamSessionID, userID, now).
+		Updates(map[string]any{"expires_at": expiresAt, "updated_at": now})
 	if result.Error != nil || result.RowsAffected == 0 {
-		return nil, common.ErrLiveNotFound
+		return nil, common.ErrLiveLeaseNotFound
 	}
 	return &vo.LiveSessionVO{Session: &session, LeaseExpiresAt: expiresAt.Unix()}, nil
 }
@@ -204,7 +210,7 @@ func (s *LiveService) ReconcileLeases() {
 	}
 	for _, session := range active {
 		if session.ExpiresAt > 0 && now.Unix() >= session.ExpiresAt {
-			if _, busiErr := s.stopForBilling(session.DockID, session.ID); busiErr != nil {
+			if _, busiErr := s.stopForBilling(session.DockID, session.ID, "session_expired"); busiErr != nil {
 				logger.ERROR("直播会话到期停止失败: "+session.ID, busiErr)
 			}
 			continue
@@ -223,7 +229,7 @@ func (s *LiveService) ReconcileLeases() {
 		if session.StopDeadline.After(now) {
 			continue
 		}
-		if _, busiErr := s.stopForBilling(session.DockID, session.ID); busiErr != nil {
+		if _, busiErr := s.stopForBilling(session.DockID, session.ID, "no_viewers"); busiErr != nil {
 			logger.ERROR("无观看者停止直播失败: "+session.ID, busiErr)
 		}
 	}
@@ -361,16 +367,33 @@ func (s *LiveService) Stop(userID int64, isAdmin bool, dockID string) *common.Bu
 	return nil
 }
 
-func (s *LiveService) stopForBilling(dockID, streamSessionID string) (*model.DeviceCommandLog, *common.BusiError) {
+func (s *LiveService) stopForBilling(dockID, streamSessionID string, reason string) (*model.DeviceCommandLog, *common.BusiError) {
+	if reason == "" {
+		reason = "no_balance"
+	}
+	// Transition first with a conditional update. This is the idempotency gate
+	// for concurrent billing/reconciliation workers: only the worker that moves
+	// the session into stopping is allowed to publish video.stop_stream.
+	updates := map[string]any{"phase": "stopping", "stop_reason": reason}
+	if reason == "no_balance" {
+		updates["error_code"] = "TRAFFIC_NOT_ENOUGH"
+	}
+	result := common.DB.Model(&model.LiveSession{}).
+		Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting')", streamSessionID, dockID).
+		Updates(updates)
+	if result.Error != nil {
+		return nil, common.ErrInternal
+	}
+	if result.RowsAffected == 0 {
+		return nil, nil
+	}
 	params := map[string]any{
 		"streamSessionId": streamSessionID,
-		"reason":          "no_balance",
+		"reason":          reason,
 	}
 	cmd, err := DefaultCommandService.DispatchToDock(dockID, "video.stop_stream", params)
 	if err != nil {
-		return nil, common.ErrInternal
-	}
-	if err := common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting')", streamSessionID, dockID).Updates(map[string]any{"phase": "stopping", "stop_reason": "no_balance"}).Error; err != nil {
+		_ = common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase = 'stopping'", streamSessionID, dockID).Updates(map[string]any{"error_code": "STOP_COMMAND_PUBLISH_FAILED"}).Error
 		return nil, common.ErrInternal
 	}
 	return cmd, nil
@@ -398,7 +421,7 @@ func (s *LiveService) GetPlayURL(userID int64, isAdmin bool, dockID, sessionID s
 		return nil, busiErr
 	}
 	var session model.LiveSession
-	query := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("dock_id = ? AND phase = 'streaming'", dockID)
+	query := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("dock_id = ? AND phase = 'streaming' AND (expires_at = 0 OR expires_at > ?)", dockID, time.Now().Unix())
 	if sessionID != "" {
 		query = query.Where("id = ?", sessionID)
 	} else {
@@ -440,6 +463,7 @@ func (s *LiveService) OnCommandAck(cmd *model.DeviceCommandLog, accepted bool, r
 	}
 	var params struct {
 		StreamSessionID string `json:"streamSessionId"`
+		Reason          string `json:"reason"`
 	}
 	if err := json.Unmarshal([]byte(cmd.Params), &params); err != nil || params.StreamSessionID == "" {
 		logger.ERROR("解析直播指令参数失败", err)
@@ -453,7 +477,19 @@ func (s *LiveService) OnCommandAck(cmd *model.DeviceCommandLog, accepted bool, r
 	if cmd.CommandType == "video.start_stream" {
 		updates["phase"] = "failed"
 	} else {
-		updates["phase"] = "streaming"
+		// A stop rejection means the stream is still active for ordinary
+		// command failures. SESSION_NOT_FOUND is different: the edge has
+		// already discarded this session (typically after a mock restart), so
+		// restoring streaming would make expiry reconciliation send stop forever.
+		if resultCode == "SESSION_NOT_FOUND" || params.Reason == "session_expired" {
+			now := time.Now()
+			updates["phase"] = "stopped"
+			updates["device_streaming"] = false
+			updates["cloud_online"] = false
+			updates["stopped_at"] = now
+		} else {
+			updates["phase"] = "streaming"
+		}
 	}
 	if err := common.DB.Model(&model.LiveSession{}).
 		Where("id = ? AND dock_id = ?", params.StreamSessionID, cmd.DockID).
@@ -484,10 +520,10 @@ func (s *LiveService) OnVideoState(dockID string, streaming bool, streamSessionI
 	// 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)
+	if result := common.DB.Select("id").Where("device_event_id = ? AND id <> ?", eventID, streamSessionID).Limit(1).Find(&eventOwner); result.Error != nil {
 		return false
-	} else if !errors.Is(err, gorm.ErrRecordNotFound) {
+	} else if result.RowsAffected > 0 {
+		logger.WARN("忽略已被其他直播会话使用的视频状态事件", eventID, streamSessionID, eventOwner.ID)
 		return false
 	}
 	updates := map[string]any{