package service

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"strconv"
	"time"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"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{}

var errLiveStopping = errors.New("live session is stopping")

func pushURLHash(pushURL string) string {
	sum := sha256.Sum256([]byte(pushURL))
	return hex.EncodeToString(sum[:])
}

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
		}
		// 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')", 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
			}
			streamID, err := tool.NextID()
			if err != nil {
				return err
			}
			expiresAt := sessionExpiry(time.Now())
			streamName := strconv.FormatInt(streamID, 10)
			// pushUrl 的签名有效期与设备硬停止时间保持一致，保证整个会话期间
			// 重试下发的同一个 pushUrl 仍然有效（协议 §7.2）。
			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, PushURLHash: pushURLHash(credentials.PushURL), 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.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 {
		if errors.Is(err, errLiveStopping) {
			return nil, common.ErrLiveStopping
		}
		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 = ? 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)
	// 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.ErrLiveLeaseNotFound
	}
	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
}

// stopForceAfter 是 stopping 的强制收敛期限。设备可能因为推流子进程卡死而一直
// 上报 streaming=true，必须在有限时间内把会话判定为 stopped，否则
// uk_live_session_active_dock 会让该机巢再也无法开启新直播。
const stopForceAfter = 120 * time.Second

func stoppingDeadline() time.Time {
	return time.Now().Add(stopForceAfter)
}

// sessionExpiry 返回会话的绝对过期时间。它有两个用途：
//  1. 作为 video.start_stream 的 expiresAt 下发给设备（设备侧硬停止时间）；
//  2. 作为推流/播放地址的签名有效期上限。
//
// 协议不允许在会话中途延长 expiresAt（重发 start_stream 会因参数摘要变化被
// 按 COMMAND_ID_CONFLICT 拒绝），所以这里必须在创建时一次性给足。
// “无人观看就停”由观看租约 + 宽限期控制，会话在有人持续心跳时会一直保持到
// 这个上限。取值以 max-session-seconds 为准，未配置时退回 auth-expire-seconds。
func sessionExpiry(now time.Time) time.Time {
	seconds := common.AppConf.Live.MaxSessionSeconds
	if seconds <= 0 {
		seconds = common.AppConf.Live.AuthExpireSeconds
	}
	if seconds <= 0 {
		seconds = 7200
	}
	return now.Add(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, "session_expired"); 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, "no_viewers"); busiErr != nil {
			logger.ERROR("无观看者停止直播失败: "+session.ID, busiErr)
		}
	}
}

func (s *LiveService) ReconcileCloudSessions() {
	provider, busiErr := s.adapter()
	if busiErr != nil {
		return
	}
	var sessions []model.LiveSession
	if err := common.DB.Where("phase IN ('starting','streaming','reconnecting','stopping')").Find(&sessions).Error; err != nil {
		logger.ERROR("扫描直播云端对账会话失败", err)
		return
	}
	now := time.Now()
	for i := range sessions {
		s.reconcileCloudSession(provider, &sessions[i], now)
	}
}

func (s *LiveService) reconcileCloudSession(provider liveprovider.Adapter, session *model.LiveSession, now time.Time) {
	online, err := provider.QueryOnline(session.StreamName)
	updates := map[string]any{"cloud_checked_at": now}
	if err != nil {
		updates["error_code"] = "PROVIDER_ONLINE_QUERY_FAILED"
		_ = common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", session.ID, session.DockID).Updates(updates).Error
		return
	}
	updates["cloud_online"] = online.Online
	if session.DeviceStreaming && online.Online {
		updates["cloud_confirmed_at"] = now
	}
	if provider.Provider() == "fake" && session.Phase == "stopping" && !session.DeviceStreaming {
		updates["cloud_online"] = false
		updates["phase"] = "stopped"
		updates["stopped_at"] = now
	}
	switch session.Phase {
	case "starting", "reconnecting":
		if session.DeviceStreaming && online.Online {
			updates["phase"] = "streaming"
			// 只在首次进入 streaming 时记开始时间，避免重连后把计费周期起点重置
			if session.StartedAt == nil {
				updates["started_at"] = now
			}
		}
	case "streaming":
		if !session.DeviceStreaming || !online.Online {
			updates["phase"] = "reconnecting"
		}
	case "stopping":
		// 设备与云端都离线，或超过强制收敛期限 → 判定已停止。
		// 后者避免设备（推流子进程）卡死时会话永久停在 stopping。
		if (!session.DeviceStreaming && !online.Online) ||
			(session.StopDeadline != nil && now.After(*session.StopDeadline)) {
			updates["phase"] = "stopped"
			updates["stopped_at"] = now
		}
	}
	_ = common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase = ?", session.ID, session.DockID, session.Phase).Updates(updates).Error
}

// OnDeviceIdle 处理设备自述“没有会话且不在推流”（state/video 里 streamSessionId 为空
// 且 streaming=false）。dock-video-relay 只把会话存在进程内存，重启后会回到 idle，
// 此时继续等阿里云超时只会让会话卡在 reconnecting。
//
// 只对“已确认过”的会话生效（streaming/reconnecting）：starting 阶段的设备还没收到
// start_stream，上报的 idle 属于正常状态，不能误杀。
func (s *LiveService) OnDeviceIdle(dockID string) {
	now := time.Now()
	result := common.DB.Model(&model.LiveSession{}).
		Where("dock_id = ? AND phase IN ('streaming','reconnecting')", dockID).
		Updates(map[string]any{
			"phase":            "stopped",
			"stop_reason":      "device_restarted",
			"error_code":       "DEVICE_SESSION_LOST",
			"device_streaming": false,
			"cloud_online":     false,
			"stopped_at":       now,
			"updated_at":       now,
		})
	if result.Error != nil {
		logger.ERROR("处理设备空闲状态失败", result.Error)
		return
	}
	if result.RowsAffected > 0 {
		logger.WARN("设备已无直播会话，会话判定结束", dockID, result.RowsAffected)
	}
}

func (s *LiveService) OnCommandTimeout(cmd model.DeviceCommandLog) {
	var params struct {
		StreamSessionID string `json:"streamSessionId"`
	}
	if json.Unmarshal([]byte(cmd.Params), &params) != nil || params.StreamSessionID == "" {
		return
	}
	if cmd.CommandType == "video.start_stream" {
		_ = common.DB.Model(&model.LiveSession{}).
			Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting')", params.StreamSessionID, cmd.DockID).
			Updates(map[string]any{"phase": "failed", "error_code": "COMMAND_ACK_TIMEOUT"}).Error
		return
	}
	// 停止指令超时：绝不能再把会话改回活跃态（reconnecting），否则对账会重新
	// 扫描到活跃会话并再次下发停止指令，形成“停→复活→再停”的死循环。
	// 这里保持 stopping，由 stop_deadline 在有限时间内强制收敛为 stopped；
	// COALESCE 保证已设置的期限不被重复超时向后推。
	_ = common.DB.Model(&model.LiveSession{}).
		Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", params.StreamSessionID, cmd.DockID).
		Updates(map[string]any{
			"error_code":    "STOP_COMMAND_ACK_TIMEOUT",
			"stop_deadline": gorm.Expr("COALESCE(stop_deadline, ?)", stoppingDeadline()),
		}).Error
}

// 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 停止指定机巢的活动会话。sessionID 非空时必须与当前活动会话一致，
// 避免旧页面拿着已过期的会话号把新会话停掉；设备仍须根据 streamSessionId 防误停。
func (s *LiveService) Stop(userID int64, isAdmin bool, dockID, sessionID 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("created_at DESC, id DESC").First(&active).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return common.ErrLiveNotFound
		}
		return common.ErrInternal
	}
	// 调用方带了会话号就必须匹配当前活动会话，防止旧页面/旧任务停掉新会话。
	if sessionID != "" && sessionID != active.ID {
		logger.WARN("拒绝停止非当前直播会话", dockID, sessionID, active.ID)
		return common.NewBusiError(common.LiveNotFound, "该直播会话已结束或已更换")
	}
	if sessionID == "" {
		logger.WARN("停止直播未携带 streamSessionId（旧客户端兼容路径）", dockID)
	}
	// 先做条件置位抢占：只有把会话从活跃态推到 stopping 的调用者才下发停流指令。
	// 重复调用（前端重试、多人同时点停止、对账任务并发）只会幂等返回，不会重复下发。
	result := common.DB.Model(&model.LiveSession{}).
		Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting')", active.ID, dockID).
		Updates(map[string]any{"phase": "stopping", "stop_reason": "manual", "stop_deadline": stoppingDeadline()})
	if result.Error != nil {
		return common.ErrInternal
	}
	if result.RowsAffected == 0 {
		return nil
	}
	params := map[string]any{
		"streamSessionId": active.ID,
		"reason":          "user_request",
	}
	if _, err := DefaultCommandService.DispatchToDock(dockID, "video.stop_stream", params); err != nil {
		logger.ERROR("下发停止推流指令失败", err)
		_ = common.DB.Model(&model.LiveSession{}).Where("id = ? AND phase = 'stopping'", active.ID).
			Updates(map[string]any{"error_code": "STOP_COMMAND_PUBLISH_FAILED"}).Error
		return common.ErrInternal
	}
	return nil
}

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, "stop_deadline": stoppingDeadline()}
	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":          protocolStopReason(reason),
	}
	cmd, err := DefaultCommandService.DispatchToDock(dockID, "video.stop_stream", params)
	if 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
}
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("created_at DESC, 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, sessionID string) (*vo.LivePlayURLVO, *common.BusiError) {
	provider, busiErr := s.adapter()
	if busiErr != nil {
		return nil, busiErr
	}
	var session model.LiveSession
	query := common.DB.Model(&model.LiveSession{}).
		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 {
		// 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")
	}
	// 用 Find 而不是 First：会话尚未进入 streaming 属于正常等待态，
	// 不应该被 GORM 记成 record not found 的 ERROR 日志。
	result := query.Limit(1).Find(&session)
	if result.Error != nil {
		logger.ERROR("查询直播会话失败", result.Error)
		return nil, common.ErrInternal
	}
	if result.RowsAffected == 0 {
		// 区分“还没就绪”和“已经结束”，让前端知道该继续等还是该收尾。
		var active int64
		existQuery := common.DB.Model(&model.LiveSession{}).
			Scopes(withDockFilter(userID, isAdmin)).
			Where("dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", dockID)
		if sessionID != "" {
			existQuery = existQuery.Where("id = ?", sessionID)
		}
		if err := existQuery.Count(&active).Error; err != nil {
			logger.ERROR("查询直播会话状态失败", err)
			return nil, common.ErrInternal
		}
		if active > 0 {
			return nil, common.ErrLiveNotReady
		}
		return nil, common.ErrLiveNotFound
	}
	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, "生成播放地址失败")
	}
	// 同时返回 HLS / FLV / RTMP：HLS 兼容性最好但延迟高，实时监控用 FLV
	// （HTTP-FLV）延迟可压到 1~3 秒。playUrl 仍保持 HLS，避免破坏旧前端。
	return &vo.LivePlayURLVO{
		StreamName:  session.StreamName,
		PlayURL:     urls.HLS,
		PlayURLHLS:  urls.HLS,
		PlayURLFLV:  urls.FLV,
		PlayURLRTMP: urls.RTMP,
		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"`
		Reason          string `json:"reason"`
	}
	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 {
		// 设备已经不认这个会话（工控机/视频服务重启后回到 idle，或会话已被替换）：
		// 继续重试停止毫无意义，直接判定结束，释放活动会话唯一约束。
		// SESSION_NOT_FOUND 是 mock 的码，VIDEO_SESSION_MISMATCH 是协议 §5.2.4 的码。
		if resultCode == "SESSION_NOT_FOUND" || resultCode == "VIDEO_SESSION_MISMATCH" {
			now := time.Now()
			updates["phase"] = "stopped"
			updates["device_streaming"] = false
			updates["cloud_online"] = false
			updates["stopped_at"] = now
		} else {
			// 一般停止失败（设备仍在推流等）保持 stopping，由 stop_deadline 强制收敛。
			// 绝不能改回 streaming，否则会与停流对账形成死循环。
			updates["phase"] = "stopping"
		}
	}
	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 ordered state from the matching non-terminal stream session.
// sequence 是幂等去重用的单调序号（协议无 version，由 state/video 快照的
// updatedAt 毫秒充当），eventId 是派生的 <streamSessionId>-<updatedAt>，
// 两者仍旧落在 device_state_version / device_event_id 两个既有列上以兼容历史数据。
func (s *LiveService) OnVideoState(dockID string, streaming bool, streamSessionID, eventID string, sequence, updatedAt, bitrateBps int64) bool {
	if streamSessionID == "" || eventID == "" || sequence <= 0 || updatedAt <= 0 {
		logger.WARN("视频状态缺少顺序元数据（OnVideoState）", dockID, streamSessionID, eventID, sequence, updatedAt)
		return false
	}
	now := time.Now()
	if abs(now.UnixMilli()-updatedAt) > int64((5*time.Minute)/time.Millisecond) {
		// updatedAt 单位不一致（秒 vs 毫秒）或设备时钟偏差会走到这里。
		logger.WARN("忽略过期视频状态", dockID, streamSessionID, "nowMs", now.UnixMilli(), "updatedAt", updatedAt)
		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 {
		logger.WARN("视频状态对应的直播会话不存在或已结束", dockID, streamSessionID, err)
		return false
	}
	// 重复投递由序号是否前进判定；eventId 只用于“跨会话归属”校验，
	// 因为协议 §7.8 的 state/video 是快照，requestId 可能在同一会话内被复用。
	if sequence <= session.DeviceStateVersion || updatedAt < session.DeviceUpdatedAt {
		logger.WARN("忽略乱序或重复的视频状态", dockID, streamSessionID,
			"sequence", sequence, "curSequence", session.DeviceStateVersion,
			"eventId", eventID, "curEventId", session.DeviceEventID,
			"updatedAt", updatedAt, "curUpdatedAt", session.DeviceUpdatedAt)
		return false
	}
	// eventId 由 <sessionId>-<updatedAt> 派生，天然唯一；这里只防止同一事件被
	// 陈旧会话重复占用（device_event_id 有唯一索引 uk_live_session_device_event）。
	var eventOwner model.LiveSession
	if result := common.DB.Select("id").Where("device_event_id = ? AND id <> ?", eventID, streamSessionID).Limit(1).Find(&eventOwner); result.Error != nil {
		logger.ERROR("校验视频状态事件唯一性失败", result.Error)
		return false
	} else if result.RowsAffected > 0 {
		logger.WARN("忽略已被其他直播会话使用的视频状态事件", eventID, streamSessionID, eventOwner.ID)
		return false
	}
	updates := map[string]any{
		"device_state_version": sequence,
		"device_event_id":      eventID,
		"device_updated_at":    updatedAt,
		"device_streaming":     streaming,
		"updated_at":           now,
	}
	// 协议 §7.8：bitrateBps 是设备侧 expectedBitrateBps（平均码率），作为计费口径。
	// 只在推流且有有效值时写入，避免停止时把计费码率清零。
	if streaming && bitrateBps > 0 {
		updates["billing_bitrate_bps"] = bitrateBps
	}
	if session.Phase == "stopping" {
		// 已下发扬停指令：设备可能仍在上报 streaming=true（或保留了停止后的会话号），
		// 绝不能把 phase 改回 streaming，否则会和停流对账来回拉扯成死循环。
		// 这里只记录设备侧事实，phase 交给对账任务收敛到 stopped。
		if streaming {
			updates["error_code"] = "STOP_IN_PROGRESS"
		} else {
			updates["error_code"] = "DEVICE_STREAM_OFFLINE"
		}
	} else if streaming {
		provider, busiErr := s.adapter()
		if busiErr != nil {
			return false
		}
		online, queryErr := provider.QueryOnline(session.StreamName)
		if queryErr != nil || !online.Online {
			updates["error_code"] = "PROVIDER_ONLINE_UNVERIFIED"
			result := common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND device_state_version < ?", streamSessionID, dockID, sequence).Updates(updates)
			if result.Error != nil {
				return false
			}
			return true
		}
		updates["phase"] = "streaming"
		// 只在首次进入 streaming 时记开始时间，避免重连后把计费周期起点重置
		if session.StartedAt == nil {
			updates["started_at"] = now
		}
		updates["cloud_confirmed_at"] = now
	} else {
		updates["phase"] = "reconnecting"
		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, sequence).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 result.RowsAffected == 1
}

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

// protocolStopReason 把内部停止原因映射为协议 §5.2.4 允许的取值，
// 避免工控机按 INVALID_PARAM 拒绝停止命令；真实原因仍记录在 live_session.stop_reason。
func protocolStopReason(reason string) string {
	switch reason {
	case "no_viewers", "user_request", "admin_stop", "session_replaced":
		return reason
	default:
		return "user_request"
	}
}