City-Lifeline-Cockpit/src/modules/cockpit/services/videoLiveSession.js

/*!
 * Live-session helper adapted from city-life-line/src/utils/videoLiveSession.js.
 * Copyright (c) 2018 RuoYi — MIT License.
 * Permission is hereby granted, free of charge, to any person obtaining a copy of
 * this software and associated documentation files (the "Software"), to deal in
 * the Software without restriction, including without limitation the rights to
 * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
 * the Software, and to permit persons to whom the Software is furnished to do so,
 * subject to the following conditions:
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
 * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
 * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
 * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */
import { selectVideoMedia, VideoRequestError } from './videoApi.js'

/** One visible attempt, owned preparation before media, a bounded retry budget and cleanup observer. */
export function createVideoLiveSession(api, update, timers = globalThis) {
    let generation = 0, current = null, cleanupTimer, cleanupChecks = 0, disposed = false
    const releases = new Set()
    const cleanupMessage = '会话释放未确认，等待服务端回收；可重试清理'
    // Retry policy: five automatic attempts after the first normal play, spaced 2/4/8/16/30s.
    const retryDelays = [2000, 4000, 8000, 16000, 30000]
    const stablePlaybackMs = 30000
    // VideoRequestError carries `category`; the player and accepted errors use `code`.
    const unrecoverable = new Set(['DEVICE_NOT_FOUND', 'VIDEO_UNAVAILABLE', 'SESSION_FORBIDDEN', 'SESSION_EXPIRED',
        'INVALID_RESPONSE', 'NOT_CONFIGURED', 'INVALID_REQUEST', 'UNSUPPORTED_MEDIA', 'VIDEO_URL_INVALID', 'VIDEO_UNSUPPORTED'])
    let retryBudget = retryDelays.length, retryTimer, retryAttempt = 0, stableTimer
    const publish = (patch) => { if (!disposed) update(patch) }
    const reportCleanup = () => publish({ cleanupError: releases.size ? cleanupMessage : '' })
    const active = (attempt) => !disposed && current === attempt && attempt.version === generation
    const clear = (attempt) => {
        if (!attempt) return
        timers.clearTimeout(attempt.startTimer); timers.clearTimeout(attempt.readyTimer); timers.clearInterval(attempt.heartbeat)
    }
    const cancelRetry = () => {
        timers.clearTimeout(retryTimer); retryTimer = undefined
        timers.clearTimeout(stableTimer); stableTimer = undefined
    }
    function scheduleCleanup() {
        if (disposed || cleanupTimer || !releases.size || cleanupChecks >= 40) return
        cleanupTimer = timers.setTimeout(async () => {
            cleanupTimer = null; cleanupChecks += 1
            for (const session of [...releases]) {
                try {
                    const result = await api.status(session)
                    if (result.sessionId === session.sessionId && ['STOPPED', 'FAILED'].includes(result.status)) {
                        session.released = true; releases.delete(session)
                    }
                } catch { /* Unknown status is not release evidence. */ }
            }
            reportCleanup(); scheduleCleanup()
        }, 3000)
    }
    async function release(session) {
        if (!session || session.released) return true
        if (session.releasing) return session.releasing
        releases.add(session)
        session.releasing = Promise.resolve().then(async () => {
            try {
                const result = await api.stop(session)
                if (result.sessionId !== session.sessionId || !['STOPPED', 'FAILED'].includes(result.status)) throw new Error('Cleanup not confirmed')
                session.released = true; releases.delete(session); return true
            } catch {
                if (session.released) return true
                publish({ cleanupError: cleanupMessage }); scheduleCleanup(); return false
            } finally { session.releasing = null }
        })
        return session.releasing
    }
    async function stop() {
        ++generation
        cancelRetry()
        const attempt = current; current = null; clear(attempt)
        publish({ media: null, phase: 'idle' })
        // An already released identity must never re-enter the pending set: release() returns early
        // for it, so the set would never shrink again and every later start would be refused.
        if (attempt?.identity && !attempt.identity.released) releases.add(attempt.identity)
        cleanupChecks = 0
        for (const session of [...releases]) await release(session)
        reportCleanup()
        return releases.size === 0
    }
    /** Only a temporary failure with budget left is retried; capability/ownership errors are final. */
    function shouldRetry(error) {
        if (disposed || !current || releases.size || retryBudget <= 0) return false
        // VideoRequestError carries `category`; accepted errors use `code`.
        const reason = error?.category || error?.code
        return !unrecoverable.has(reason)
    }

    /** Spends one attempt of the budget and schedules the next attempt after its own delay. */
    function scheduleRetry(deviceCode, message) {
        const attempt = current
        // Spend the budget here, before the timer exists, so the last retry cannot schedule another.
        retryBudget -= 1
        retryAttempt += 1
        const delay = retryDelays[retryDelays.length - retryBudget - 1] ?? retryDelays[retryDelays.length - 1]
        publish({ phase: 'retrying', error: message, retryAttempt, retryDelayMs: delay, retriesLeft: retryBudget })
        retryTimer = timers.setTimeout(() => {
            retryTimer = undefined
            if (!active(attempt) || releases.size) return
            void begin(deviceCode)
        }, delay)
    }
    async function fail(message, media) {
        if (!current || (media && current.media !== media)) return
        const error = message instanceof Error ? message : Object.assign(new Error(message || '视频播放失败'), {})
        const deviceCode = current.deviceCode
        publish({ error: error.message || '视频播放失败' })
        const attempt = current
        const retryable = shouldRetry(error)
        const stopping = stop(), version = generation
        const released = await stopping
        if (disposed || version !== generation) return
        if (!retryable || !released) {
            // Running the budget down is actionable information, distinct from a permanent capability error.
            const exhausted = retryAttempt >= retryDelays.length && !unrecoverable.has(error?.category || error?.code)
            const reason = exhausted ? `${error.message}（已自动重试 ${retryDelays.length} 次仍失败）` : error.message
            publish({ phase: 'failed', error: reason, retryAttempt: 0, retryDelayMs: 0, retriesLeft: 0 })
            return
        }
        // `released` proved the old attempt is fully gone; only its cancellation token is restored so the
        // scheduled retry still belongs to this viewing session. A fresh attempt is built by begin().
        current = { version, deviceCode, identity: null, media: null }
        scheduleRetry(deviceCode, error.message)
    }
    function ready(media) {
        if (!current?.media || current.identity?.released || (media && current.media !== media)) return
        timers.clearTimeout(current.readyTimer)
        publish({ phase: 'playing' })
        // A retry budget is earned only after real, sustained playback.
        timers.clearTimeout(stableTimer)
        const attempt = current
        stableTimer = timers.setTimeout(() => {
            if (!active(attempt)) return
            retryBudget = retryDelays.length; retryAttempt = 0
        }, stablePlaybackMs)
    }
    async function begin(deviceCode) {
        if (disposed) return
        if (releases.size) {
            publish({ phase: 'failed', error: '上次清理结果尚未确认，暂不能重复建流' }); return
        }
        const version = generation
        const attempt = { version, deviceCode, identity: null, media: null }
        current = attempt
        publish({ phase: 'starting', media: null, error: '', cleanupError: '' })
        attempt.startTimer = timers.setTimeout(() => { if (active(attempt)) void fail('视频建立超时') }, 35000)
        try {
            const identity = await api.prepare(deviceCode)
            if (typeof identity.sessionId !== 'string' || !/^[A-Za-z0-9_-]{1,128}$/.test(identity.sessionId)
                || typeof identity.sessionToken !== 'string' || !identity.sessionToken)
                throw new VideoRequestError('INVALID_RESPONSE', '媒体启动前未取得完整观看身份')
            attempt.identity = { sessionId: identity.sessionId, sessionToken: identity.sessionToken, deviceCode, mode: 'LIVE' }
            if (!active(attempt)) { await release(attempt.identity); return }
            if (identity.deviceCode !== deviceCode || identity.mode !== 'LIVE' || identity.status !== 'PREPARED')
                throw new VideoRequestError('INVALID_RESPONSE', '观看身份与当前设备不匹配')
            let data = await api.start(attempt.identity)
            while (active(attempt) && data.status === 'STARTING') {
                if (data.sessionId !== attempt.identity.sessionId) throw new VideoRequestError('INVALID_RESPONSE', '返回的观看会话不匹配')
                await new Promise((resolve) => timers.setTimeout(resolve, 1000))
                if (!active(attempt)) break
                data = await api.status(attempt.identity)
            }
            if (!active(attempt)) { await release(attempt.identity); return }
            if (data.sessionId !== attempt.identity.sessionId) throw new VideoRequestError('INVALID_RESPONSE', '返回的观看会话不匹配')
            if (['STOPPED', 'FAILED'].includes(data.status)) {
                attempt.identity.released = true; clear(attempt); current = null
                publish({ media: null, phase: 'failed', error: '观看会话未建立或已停止' }); return
            }
            if (data.mode !== 'LIVE') throw new VideoRequestError('INVALID_RESPONSE', '返回的播放会话与当前设备不匹配')
            // 后端 preparedResult 只回 sessionId/status/mode/media；deviceCode 可选，只有在确实回了且不一致时才算不匹配。
            if (data.deviceCode !== undefined && data.deviceCode !== deviceCode)
                throw new VideoRequestError('INVALID_RESPONSE', '返回的播放会话与当前设备不匹配')
            if (data.status !== 'ACTIVE') throw new VideoRequestError('SESSION_PENDING', '媒体尚未就绪，结束本次观看')
            attempt.media = selectVideoMedia(data.media)
            timers.clearTimeout(attempt.startTimer)
            publish({ phase: 'connecting', media: attempt.media })
            attempt.readyTimer = timers.setTimeout(() => { if (active(attempt)) void fail('播放器就绪超时') }, 20000)
            let beating = false
            attempt.heartbeat = timers.setInterval(async () => {
                if (beating || !active(attempt)) return
                beating = true
                try { await api.heartbeat(attempt.identity) }
                catch { if (active(attempt)) await fail('播放会话已断开，请重新播放') }
                finally { beating = false }
            }, 30000)
        } catch (error) {
            if (active(attempt)) await fail(error)
            else if (attempt.identity) await release(attempt.identity)
        } finally {
            timers.clearTimeout(attempt.startTimer)
            reportCleanup()
        }
    }
    async function start(deviceCode) {
        const stopping = stop(), version = generation
        await stopping
        if (disposed || version !== generation) return
        // A manual replay starts a fresh budget; automatic retries never call this.
        retryBudget = retryDelays.length; retryAttempt = 0
        publish({ retryAttempt: 0, retryDelayMs: 0, retriesLeft: retryDelays.length })
        await begin(deviceCode)
    }
    async function dispose() { await stop(); disposed = true; timers.clearTimeout(cleanupTimer) }
    return { start, stop, dispose, ready, fail }
}