laic-backend/service/live_service.go

package service

import (
	"encoding/json"
	"errors"
	"strconv"
	"time"

	"gorm.io/gorm"

	"laic-backend/cache"
	"laic-backend/common"
	liveprovider "laic-backend/live"
	"laic-backend/logger"
	"laic-backend/model"
	"laic-backend/tool"
	"laic-backend/vo"
)

type LiveService struct {
	provider liveprovider.Adapter
}

var DefaultLiveService = &LiveService{}

func InitLiveProvider(conf common.Live) error {
	provider, err := liveprovider.NewAdapter(conf)
	if err != nil {
		return err
	}
	DefaultLiveService.provider = provider
	return nil
}

func (s *LiveService) adapter() (liveprovider.Adapter, *common.BusiError) {
	if s.provider == nil || s.provider.Provider() == "disabled" {
		return nil, common.NewBusiError(common.LiveProviderDisabled, "直播服务未配置")
	}
	return s.provider, nil
}

func (s *LiveService) Join(userID int64, isAdmin bool, dockID string, req *vo.LiveSessionCreateReq) (*vo.LiveSessionVO, *common.BusiError) {
	provider, busiErr := s.adapter()
	if busiErr != nil {
		return nil, busiErr
	}
	var dock model.Dock
	if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("dock_id = ?", dockID).First(&dock).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, common.ErrDockNotFound
		}
		return nil, common.ErrInternal
	}
	online, _ := common.SetMemberExists(cache.OnlineDockSetKey, dockID)
	if !online {
		return nil, common.ErrDockOffline
	}

	var session model.LiveSession
	var lease model.LiveViewerLease
	var startParams map[string]any
	err := common.DB.Transaction(func(tx *gorm.DB) error {
		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").First(&session).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
			return err
		}
		if session.ID == "" {
			maxBitrate := req.MaxBitrateBps
			if maxBitrate <= 0 {
				maxBitrate = 1500000
			}
			streamID, err := tool.NextID()
			if err != nil {
				return err
			}
			expireSec := common.AppConf.Live.AuthExpireSeconds
			if expireSec <= 0 {
				expireSec = 7200
			}
			expiresAt := time.Now().Add(time.Duration(expireSec) * time.Second)
			streamName := strconv.FormatInt(streamID, 10)
			credentials, err := provider.CreateStream(liveprovider.StreamRequest{StreamName: streamName, ExpiresAt: expiresAt, MaxBitrateBps: maxBitrate})
			if err != nil {
				return err
			}
			now := time.Now()
			session = model.LiveSession{ID: streamName, DockID: dockID, Provider: credentials.Provider, StreamName: streamName, ExpiresAt: expiresAt.Unix(), MaxBitrateBps: maxBitrate, Phase: "starting", RequestedBy: userID, CreatedAt: now, UpdatedAt: now}
			if err := tx.Create(&session).Error; err != nil {
				return err
			}
			startParams = map[string]any{"provider": credentials.Provider, "streamSessionId": session.ID, "pushUrl": credentials.PushURL, "expiresAt": credentials.ExpiresAt.UnixMilli(), "maxBitrateBps": maxBitrate}
		}
		now := time.Now()
		leaseID, err := tool.NextID()
		if err != nil {
			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
	})
	if err != nil {
		logger.ERROR("创建直播观看租约失败", err)
		return nil, common.ErrInternal
	}
	if startParams != nil {
		if _, err := DefaultCommandService.DispatchToDock(dockID, "video.start_stream", startParams); err != nil {
			logger.ERROR("下发推流指令失败", err)
			_ = common.DB.Model(&model.LiveSession{}).Where("id = ?", session.ID).Updates(map[string]any{"phase": "failed", "error_code": "MQTT_PUBLISH_FAILED"}).Error
			return nil, common.ErrInternal
		}
	}
	return &vo.LiveSessionVO{Session: &session, LeaseExpiresAt: lease.ExpiresAt.Unix()}, nil
}

func leaseSeconds() int {
	seconds := common.AppConf.Live.ViewerLeaseSeconds
	if seconds <= 0 {
		return 30
	}
	return seconds
}

func (s *LiveService) GetSession(userID int64, isAdmin bool, dockID, streamSessionID string) (*model.LiveSession, *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 errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, common.ErrLiveNotFound
		}
		return nil, common.ErrInternal
	}
	return &session, nil
}
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 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()})
	if result.Error != nil || result.RowsAffected == 0 {
		return nil, common.ErrLiveNotFound
	}
	return &vo.LiveSessionVO{Session: &session, LeaseExpiresAt: expiresAt.Unix()}, nil
}

func (s *LiveService) Leave(userID int64, isAdmin bool, dockID, streamSessionID string) *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 errors.Is(err, gorm.ErrRecordNotFound) {
			return common.ErrLiveNotFound
		}
		return common.ErrInternal
	}
	now := time.Now()
	if err := common.DB.Model(&model.LiveViewerLease{}).Where("stream_session_id = ? AND viewer_id = ? AND released_at IS NULL", streamSessionID, userID).Updates(map[string]any{"released_at": now, "updated_at": now}).Error; err != nil {
		return common.ErrInternal
	}
	return nil
}

func stopGraceSeconds() time.Duration {
	seconds := common.AppConf.Live.StopGraceSeconds
	if seconds <= 0 {
		seconds = 15
	}
	return time.Duration(seconds) * time.Second
}

// ReconcileLeases expires stale viewer leases and stops unviewed sessions after the grace period.
func (s *LiveService) ReconcileLeases() {
	now := time.Now()
	if err := common.DB.Model(&model.LiveViewerLease{}).Where("released_at IS NULL AND expires_at <= ?", now).Updates(map[string]any{"released_at": now, "updated_at": now}).Error; err != nil {
		logger.ERROR("清理过期直播租约失败", err)
		return
	}

	var active []model.LiveSession
	if err := common.DB.Where("phase IN ('starting','streaming','reconnecting')").Find(&active).Error; err != nil {
		logger.ERROR("扫描活动直播会话失败", err)
		return
	}
	for _, session := range active {
		if session.ExpiresAt > 0 && now.Unix() >= session.ExpiresAt {
			if _, busiErr := s.stopForBilling(session.DockID, session.ID); busiErr != nil {
				logger.ERROR("直播会话到期停止失败: "+session.ID, busiErr)
			}
			continue
		}
		var viewers int64
		if err := common.DB.Model(&model.LiveViewerLease{}).Where("stream_session_id = ? AND released_at IS NULL AND expires_at > ?", session.ID, now).Count(&viewers).Error; err != nil || viewers > 0 {
			if session.StopDeadline != nil {
				_ = common.DB.Model(&model.LiveSession{}).Where("id = ?", session.ID).Update("stop_deadline", nil).Error
			}
			continue
		}
		if session.StopDeadline == nil {
			_ = common.DB.Model(&model.LiveSession{}).Where("id = ? AND stop_deadline IS NULL", session.ID).Update("stop_deadline", now.Add(stopGraceSeconds())).Error
			continue
		}
		if session.StopDeadline.After(now) {
			continue
		}
		if _, busiErr := s.stopForBilling(session.DockID, session.ID); busiErr != nil {
			logger.ERROR("无观看者停止直播失败: "+session.ID, busiErr)
		}
	}
}

// Start 兼容旧启动接口，统一复用创建或加入会话的租约流程。
func (s *LiveService) Start(userID int64, isAdmin bool, dockID string, req *vo.LiveStartReq) (*model.LiveSession, *common.BusiError) {
	result, busiErr := s.Join(userID, isAdmin, dockID, &vo.LiveSessionCreateReq{MaxBitrateBps: req.MaxBitrateBps})
	if busiErr != nil {
		return nil, busiErr
	}
	return result.Session, nil
}

// Stop 停止指定机巢最新活动会话；设备必须根据 streamSessionId 防止误停新会话。
func (s *LiveService) Stop(userID int64, isAdmin bool, dockID string) *common.BusiError {
	var dock model.Dock
	if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("dock_id = ?", dockID).First(&dock).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return common.ErrDockNotFound
		}
		return common.ErrInternal
	}
	var active model.LiveSession
	if err := common.DB.Where("dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", dockID).Order("id DESC").First(&active).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return common.ErrLiveNotFound
		}
		return common.ErrInternal
	}
	params := map[string]any{
		"streamSessionId": active.ID,
		"reason":          "user_request",
	}
	if _, err := DefaultCommandService.DispatchToDock(dockID, "video.stop_stream", params); err != nil {
		logger.ERROR("下发停止推流指令失败", err)
		return common.ErrInternal
	}
	if err := common.DB.Model(&model.LiveSession{}).Where("id = ? AND phase IN ('starting','streaming','reconnecting')", active.ID).Updates(map[string]any{"phase": "stopping", "stop_reason": "manual"}).Error; err != nil {
		return common.ErrInternal
	}
	return nil
}

func (s *LiveService) stopForBilling(dockID, streamSessionID string) (*model.DeviceCommandLog, *common.BusiError) {
	params := map[string]any{
		"streamSessionId": streamSessionID,
		"reason":          "no_balance",
	}
	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 {
		return nil, common.ErrInternal
	}
	return cmd, nil
}
func (s *LiveService) GetPage(userID int64, isAdmin bool, req *vo.LivePageReq) (*common.PageResponse[model.LiveSession], *common.BusiError) {
	db := common.DB.Model(&model.LiveSession{}).Scopes(withDockFilter(userID, isAdmin))
	if req.Phase != "" {
		db = db.Where("phase = ?", req.Phase)
	}
	var total int64
	if err := db.Count(&total).Error; err != nil {
		return nil, common.ErrInternal
	}
	var list []model.LiveSession
	if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil {
		return nil, common.ErrInternal
	}
	return common.Page(req.Pagination, total, list), nil
}

// GetPlayURL only returns a URL after both edge and provider confirmation are represented by streaming.
func (s *LiveService) GetPlayURL(userID int64, isAdmin bool, dockID string) (*vo.LivePlayURLVO, *common.BusiError) {
	provider, busiErr := s.adapter()
	if busiErr != nil {
		return nil, busiErr
	}
	var session model.LiveSession
	if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("dock_id = ? AND phase = 'streaming'", dockID).Order("id DESC").First(&session).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, common.ErrLiveNotFound
		}
		return nil, common.ErrInternal
	}
	var lease model.LiveViewerLease
	if err := common.DB.Where("stream_session_id = ? AND viewer_id = ? AND released_at IS NULL AND expires_at > ?", session.ID, userID, time.Now()).First(&lease).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, common.ErrLiveLeaseNotFound
		}
		return nil, common.ErrInternal
	}
	ttl := common.AppConf.Live.PlayURLTTLSeconds
	if ttl <= 0 || ttl > 300 {
		ttl = 300
	}
	urls, err := provider.CreatePlayURLs(liveprovider.PlayRequest{StreamName: session.StreamName, ExpiresAt: time.Now().Add(time.Duration(ttl) * time.Second)})
	if err != nil {
		return nil, common.NewBusiError(common.LiveProviderInvalid, "生成播放地址失败")
	}
	return &vo.LivePlayURLVO{StreamName: session.StreamName, PlayURL: urls.HLS, ExpiresAt: urls.ExpiresAt.Unix()}, nil
}

func (s *LiveService) OnCommandAck(cmd *model.DeviceCommandLog, accepted bool, resultCode string) {
	if cmd.CommandType != "video.start_stream" && cmd.CommandType != "video.stop_stream" {
		return
	}
	var params struct {
		StreamSessionID string `json:"streamSessionId"`
	}
	if err := json.Unmarshal([]byte(cmd.Params), &params); err != nil || params.StreamSessionID == "" {
		logger.ERROR("解析直播指令参数失败", err)
		return
	}
	if accepted {
		return
	}

	updates := map[string]any{"error_code": resultCode}
	if cmd.CommandType == "video.start_stream" {
		updates["phase"] = "failed"
	} else {
		updates["phase"] = "streaming"
	}
	if err := common.DB.Model(&model.LiveSession{}).
		Where("id = ? AND dock_id = ?", params.StreamSessionID, cmd.DockID).
		Updates(updates).Error; err != nil {
		logger.ERROR("更新直播指令失败状态失败", err)
	}
}

// OnVideoState accepts only ordered state from the matching non-terminal stream session.
func (s *LiveService) OnVideoState(dockID string, streaming bool, streamSessionID, eventID string, version, updatedAt int64) bool {
	if streamSessionID == "" || eventID == "" || version <= 0 || updatedAt <= 0 {
		return false
	}
	now := time.Now()
	if abs(now.UnixMilli()-updatedAt) > int64((5*time.Minute)/time.Millisecond) {
		logger.WARN("忽略过期视频状态", dockID, streamSessionID)
		return false
	}

	var session model.LiveSession
	if err := common.DB.Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", streamSessionID, dockID).First(&session).Error; err != nil {
		return false
	}
	if version <= session.DeviceStateVersion || (session.DeviceEventID != "" && session.DeviceEventID == eventID) || updatedAt < session.DeviceUpdatedAt {
		return false
	}
	updates := map[string]any{
		"device_state_version": version,
		"device_event_id":      eventID,
		"device_updated_at":    updatedAt,
		"updated_at":           now,
	}
	if streaming {
		provider, busiErr := s.adapter()
		if busiErr != nil {
			return false
		}
		online, queryErr := provider.QueryOnline(streamSessionID)
		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
		}
		updates["phase"] = "streaming"
		updates["started_at"] = now
		updates["cloud_confirmed_at"] = now
	} else {
		updates["phase"] = "stopped"
		updates["stop_reason"] = "device"
		updates["stopped_at"] = now
	}
	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
}

func abs(value int64) int64 {
	if value < 0 {
		return -value
	}
	return value
}