import { HttpStatusCode } from '@peertube/peertube-models'
import { logger } from '@server/helpers/logger.js'
import { buildRtspSinkUrlForTargetCamera } from '@server/lib/live-videos/rtsp-url.js'
import { LiveVideoModel, LiveVideoStatus } from '@server/models/video/live-video.js'
import { CONFIG } from '@server/initializers/config.js'
import { asyncMiddleware } from '@server/middlewares/async.js'
import express from 'express'
import { readFile } from 'node:fs/promises'
import { existsSync } from 'node:fs'
import { mkdir, rm } from 'node:fs/promises'
import { join, resolve } from 'node:path'
import { setTimeout as wait } from 'node:timers/promises'
import { ChildProcessWithoutNullStreams, spawn } from 'node:child_process'
import yaml from 'js-yaml'

type StreamConfig = {
  streams?: Record<string, string | string[]>
}

type LiveSession = {
  ffmpegProcess?: ChildProcessWithoutNullStreams
  keepAliveUntil?: number
  startedAt: number
  lastAccessAt: number
  startingPromise?: Promise<void>
  restartTimer?: NodeJS.Timeout
  lastError?: string
  stopped?: boolean
}

const SESSION_IDLE_TIMEOUT_MS = 300_000
const SESSION_SWEEP_INTERVAL_MS = 20_000
const MANIFEST_WAIT_TIMEOUT_MS = 20_000
const MANIFEST_WAIT_INTERVAL_MS = 250
const LIVE_CONFIG_PATH = join(process.cwd(), 'livestream-sources.yaml')
const LIVE_WORKDIR = join(CONFIG.STORAGE.TMP_PERSISTENT_DIR, 'native-live')

const liveStreamRouter = express.Router()
const liveSessions = new Map<string, LiveSession>()

if(!CONFIG.LIVE.AUTO_START_FFMPEG_STREAMING) {
  const sweepTimer = setInterval(cleanupInactiveSessions, SESSION_SWEEP_INTERVAL_MS)
  sweepTimer.unref()
}

liveStreamRouter.get('/streams', asyncMiddleware(listStreams))
liveStreamRouter.get('/streams/:streamKey/index.m3u8', asyncMiddleware(getManifest))
liveStreamRouter.get('/streams/:streamKey/:segmentName', asyncMiddleware(getSegment))
liveStreamRouter.get('/scheduled-streams/:cameraId/index.m3u8', asyncMiddleware(getScheduledManifest))
liveStreamRouter.get('/scheduled-streams/:cameraId/:segmentName', asyncMiddleware(getScheduledSegment))

export { liveStreamRouter, initLiveStreams, startScheduledLiveSession, stopScheduledLiveSession }

async function listStreams (req: express.Request, res: express.Response) {
  const streams = await readConfiguredStreams()

  return res.json({
    data: Object.keys(streams)
  })
}

async function getManifest (req: express.Request, res: express.Response) {
  const streamKey = sanitizeStreamKey(req.params.streamKey)

  if (!streamKey) {
    return res.fail({
      status: HttpStatusCode.BAD_REQUEST_400,
      message: 'Invalid stream key.'
    })
  }

  const sourceUrl = await findStreamSourceUrl(streamKey)
  if (!sourceUrl) {
    return res.fail({
      status: HttpStatusCode.NOT_FOUND_404,
      message: `Unknown stream key "${streamKey}".`
    })
  }

  return serveManifest(res, streamKey, sourceUrl)
}

async function getScheduledManifest (req: express.Request, res: express.Response) {
  const cameraId = sanitizeStreamKey(req.params.cameraId)

  if (!cameraId) {
    return res.fail({
      status: HttpStatusCode.BAD_REQUEST_400,
      message: 'Invalid camera id.'
    })
  }

  const activeLiveVideo = await loadActiveScheduledLiveVideo(cameraId)
  if (!activeLiveVideo) {
    await stopScheduledLiveSession(cameraId)
      .catch(err => logger.error('Cannot stop stale scheduled sink stream session after expired manifest request.', { err, cameraId }))

    return res.fail({
      status: HttpStatusCode.NOT_FOUND_404,
      message: 'No live stream is currently active for this camera.'
    })
  }

  return serveManifest(res, buildScheduledStreamSessionKey(cameraId), buildRtspSinkUrlForTargetCamera(cameraId))
}

async function getSegment (req: express.Request, res: express.Response) {
  const streamKey = sanitizeStreamKey(req.params.streamKey)
  const segmentName = req.params.segmentName

  if (!streamKey) {
    return res.fail({
      status: HttpStatusCode.BAD_REQUEST_400,
      message: 'Invalid stream key.'
    })
  }

  if (!segmentName || !/^segment_[0-9]{6}\.ts$/.test(segmentName)) {
    return res.fail({
      status: HttpStatusCode.BAD_REQUEST_400,
      message: 'Invalid segment name.'
    })
  }

  const sourceUrl = await findStreamSourceUrl(streamKey)
  if (!sourceUrl) {
    return res.fail({
      status: HttpStatusCode.NOT_FOUND_404,
      message: `Unknown stream key "${streamKey}".`
    })
  }

  return serveSegment(res, streamKey, segmentName, sourceUrl)
}

async function getScheduledSegment (req: express.Request, res: express.Response) {
  const cameraId = sanitizeStreamKey(req.params.cameraId)
  const segmentName = req.params.segmentName

  if (!cameraId) {
    return res.fail({
      status: HttpStatusCode.BAD_REQUEST_400,
      message: 'Invalid camera id.'
    })
  }

  if (!segmentName || !/^segment_[0-9]{6}\.ts$/.test(segmentName)) {
    return res.fail({
      status: HttpStatusCode.BAD_REQUEST_400,
      message: 'Invalid segment name.'
    })
  }

  const activeLiveVideo = await loadActiveScheduledLiveVideo(cameraId)
  if (!activeLiveVideo) {
    await stopScheduledLiveSession(cameraId)
      .catch(err => logger.error('Cannot stop stale scheduled sink stream session after expired segment request.', { err, cameraId, segmentName }))

    return res.fail({
      status: HttpStatusCode.NOT_FOUND_404,
      message: 'No live stream is currently active for this camera.'
    })
  }

  return serveSegment(res, buildScheduledStreamSessionKey(cameraId), segmentName, buildRtspSinkUrlForTargetCamera(cameraId))
}

async function serveManifest (res: express.Response, streamKey: string, sourceUrl: string) {
  const readyToServe = await ensureLiveSession(streamKey, sourceUrl)
  if (!readyToServe) {
    return res.fail({
      status: HttpStatusCode.NOT_FOUND_404,
      message: 'No live stream is currently active for this camera.'
    })
  }

  const manifestPath = buildStreamPath(streamKey, 'index.m3u8')
  const ready = await waitForFile(manifestPath, MANIFEST_WAIT_TIMEOUT_MS)

  if (!ready) {
    const session = liveSessions.get(streamKey)

    return res.fail({
      status: HttpStatusCode.SERVICE_UNAVAILABLE_503,
      message: session?.lastError || 'Stream is starting, please retry in a few seconds.'
    })
  }

  touchSession(streamKey)
  res.setHeader('Cache-Control', 'public, max-age=1, must-revalidate')
  return res.sendFile(resolve(manifestPath))
}

async function serveSegment (res: express.Response, streamKey: string, segmentName: string, sourceUrl: string) {
  const readyToServe = await ensureLiveSession(streamKey, sourceUrl)
  if (!readyToServe) {
    return res.fail({
      status: HttpStatusCode.NOT_FOUND_404,
      message: 'No live stream is currently active for this camera.'
    })
  }

  const segmentPath = buildStreamPath(streamKey, segmentName)
  if (!existsSync(segmentPath)) {
    return res.fail({
      status: HttpStatusCode.NOT_FOUND_404,
      message: 'Segment not found.'
    })
  }

  touchSession(streamKey)
  res.setHeader('Cache-Control', 'public, max-age=30, immutable')
  return res.sendFile(resolve(segmentPath))
}

async function readConfiguredStreams () {
  if (!existsSync(LIVE_CONFIG_PATH)) {
    return {}
  }

  const raw = await readFile(LIVE_CONFIG_PATH, 'utf-8')
  const parsed = yaml.load(raw) as StreamConfig | undefined
  return parsed?.streams || {}
}

async function findStreamSourceUrl (streamKey: string) {
  const streams = await readConfiguredStreams()
  const streamValue = streams[streamKey]

  if (!streamValue) {
    return buildRtspSinkUrlForTargetCamera(streamKey)
  }

  if (typeof streamValue === 'string') return streamValue

  const firstRtsp = streamValue.find(v => typeof v === 'string' && v.startsWith('rtsp://'))
  if (firstRtsp) return firstRtsp

  return streamValue.find(v => typeof v === 'string') || buildRtspSinkUrlForTargetCamera(streamKey)
}

function sanitizeStreamKey (streamKey: string) {
  if (!streamKey) return undefined
  if (!/^[A-Za-z0-9._-]+$/.test(streamKey)) return undefined

  return streamKey
}

function buildStreamPath (streamKey: string, fileName: string) {
  return join(LIVE_WORKDIR, streamKey, fileName)
}

function buildScheduledStreamSessionKey (cameraId: string) {
  return `scheduled-${cameraId}`
}

async function startScheduledLiveSession (cameraId: string, options: { keepAliveUntil?: Date | number } = {}) {
  const streamKey = buildScheduledStreamSessionKey(cameraId)
  const keepAliveUntil = normalizeKeepAliveUntil(options.keepAliveUntil)

  return ensureLiveSession(streamKey, buildRtspSinkUrlForTargetCamera(cameraId), { keepAliveUntil })
}

async function stopScheduledLiveSession (cameraId: string) {
  return stopLiveSession(buildScheduledStreamSessionKey(cameraId))
}

function touchSession (streamKey: string) {
  const session = liveSessions.get(streamKey)
  if (!session) return

  session.lastAccessAt = Date.now()
}

async function ensureLiveSession (streamKey: string, sourceUrl: string, options: { keepAliveUntil?: number } = {}) {
  let session = liveSessions.get(streamKey)
  const now = Date.now()

  if ((options.keepAliveUntil || 0) <= now && options.keepAliveUntil !== undefined) {
    await stopLiveSession(streamKey)
    return false
  }

  if (session && (session.keepAliveUntil || 0) <= now && session.keepAliveUntil !== undefined) {
    await stopLiveSession(streamKey)
    return false
  }

  if (!session) {
    session = {
      startedAt: 0,
      lastAccessAt: Date.now(),
      lastError: undefined,
      stopped: false
    }
    liveSessions.set(streamKey, session)
  }

  session.stopped = false
  clearRestartTimer(session)

  if (options.keepAliveUntil) {
    session.keepAliveUntil = Math.max(session.keepAliveUntil || 0, options.keepAliveUntil)
  }

  if (isChildProcessAlive(session.ffmpegProcess)) {
    touchSession(streamKey)
    return true
  }

  if (session.startingPromise) {
    await session.startingPromise
    touchSession(streamKey)
    return true
  }

  session.startingPromise = startSession(streamKey, sourceUrl)
  try {
    await session.startingPromise
  } finally {
    session.startingPromise = undefined
  }

  touchSession(streamKey)
  return true
}

async function startSession (streamKey: string, sourceUrl: string) {
  const outputDirectory = join(LIVE_WORKDIR, streamKey)
  await rm(outputDirectory, { recursive: true, force: true })
  await mkdir(outputDirectory, { recursive: true })

  const manifestPath = join(outputDirectory, 'index.m3u8')
  const segmentPattern = join(outputDirectory, 'segment_%06d.ts')

  const ffmpegPath = CONFIG.LIVE.FFMPEG.PATH?.trim() || 'ffmpeg'
  const segmentDuration = Math.max(1, CONFIG.LIVE.FFMPEG.HLS_TIME)
  const hlsListSize = Math.max(1, CONFIG.LIVE.FFMPEG.HLS_LIST_SIZE)

  const ffmpegArgs = [
    '-hide_banner',
    '-loglevel', 'warning',
    '-fflags', '+genpts+igndts+discardcorrupt',
    '-rtsp_transport', 'tcp',
    '-thread_queue_size', '512',
    '-i', sourceUrl,
    '-map', '0:v:0',
    '-map', '0:a:0?',
    '-c', 'copy',
    '-f', 'hls',
    '-hls_time', String(segmentDuration),
    '-hls_list_size', String(hlsListSize),
    '-hls_delete_threshold', '2',
    '-hls_flags', 'delete_segments+independent_segments+omit_endlist',
    '-hls_segment_filename', segmentPattern,
    manifestPath
  ]

  const ffmpegProcess = spawn(ffmpegPath, ffmpegArgs, {
    stdio: [ 'ignore', 'ignore', 'pipe' ]
  })

  const session = liveSessions.get(streamKey)
  if (!session || session.stopped) {
    ffmpegProcess.kill('SIGTERM')
    return
  }

  session.ffmpegProcess = ffmpegProcess
  session.startedAt = Date.now()
  session.lastAccessAt = Date.now()
  session.lastError = undefined

  let stderrBuffer = ''
  ffmpegProcess.stderr.on('data', chunk => {
    // Keep pipe drained to avoid stalling ffmpeg and keep last lines for diagnostics.
    const line = chunk?.toString?.() || ''
    stderrBuffer = (stderrBuffer + line).slice(-4_000)
  })

  ffmpegProcess.once('error', err => {
    logger.error('Cannot start ffmpeg live session', { err, streamKey })

    const existing = liveSessions.get(streamKey)
    if (!existing) return

    existing.lastError = `Cannot start ffmpeg: ${err.message}`
  })

  ffmpegProcess.once('exit', () => {
    const existing = liveSessions.get(streamKey)
    if (!existing) return
    if (existing.ffmpegProcess !== ffmpegProcess) return

    existing.ffmpegProcess = undefined
    existing.startedAt = 0

    const trimmedError = stderrBuffer.trim()
    if (trimmedError) {
      const lines = trimmedError.split('\n')
      existing.lastError = lines.slice(-3).join(' | ')
    }

    const shouldRestart = CONFIG.LIVE.AUTO_START_FFMPEG_STREAMING ||
      ((existing.keepAliveUntil || 0) > Date.now())

    if (shouldRestart && !existing.stopped) {
      const keepAliveUntil = existing.keepAliveUntil
      clearRestartTimer(existing)
      existing.restartTimer = setTimeout(() => {
        const current = liveSessions.get(streamKey)
        if (current !== existing || current?.stopped) return

        void ensureLiveSession(streamKey, sourceUrl, { keepAliveUntil })
      }, 3_000)
    }
  })
}

async function stopLiveSession (streamKey: string) {
  const session = liveSessions.get(streamKey)
  if (!session) return false

  session.stopped = true
  session.keepAliveUntil = undefined
  session.startingPromise = undefined
  clearRestartTimer(session)

  if (session.ffmpegProcess) {
    stopSessionProcess(session.ffmpegProcess, streamKey)
  }

  liveSessions.delete(streamKey)
  return true
}

async function waitForFile (filePath: string, timeoutMs: number) {
  const end = Date.now() + timeoutMs

  while (Date.now() < end) {
    if (existsSync(filePath)) return true

    await wait(MANIFEST_WAIT_INTERVAL_MS)
  }

  return false
}

async function initLiveStreams () {
  const streams = await readConfiguredStreams()

  for (const streamKey of Object.keys(streams)) {
    const sourceUrl = await findStreamSourceUrl(streamKey)
    if (sourceUrl) {
      try {
        await ensureLiveSession(streamKey, sourceUrl)
      } catch (err) {
        logger.warn('Failed to start live stream %s', streamKey, { err })
      }
    }
  }
}

function cleanupInactiveSessions () {
  const now = Date.now()

  for (const [ streamKey, session ] of liveSessions.entries()) {
    if ((session.keepAliveUntil || 0) > now) {
      continue
    }

    if (!isChildProcessAlive(session.ffmpegProcess)) {
      liveSessions.delete(streamKey)
      continue
    }

    if (now - session.lastAccessAt < SESSION_IDLE_TIMEOUT_MS) continue
    stopSessionProcess(session.ffmpegProcess, streamKey)
    liveSessions.delete(streamKey)
  }
}

function normalizeKeepAliveUntil (keepAliveUntil?: Date | number) {
  if (keepAliveUntil instanceof Date) return keepAliveUntil.getTime()
  if (typeof keepAliveUntil === 'number' && Number.isFinite(keepAliveUntil)) return keepAliveUntil

  return undefined
}

function clearRestartTimer (session: LiveSession) {
  if (!session.restartTimer) return

  clearTimeout(session.restartTimer)
  session.restartTimer = undefined
}

function isChildProcessAlive (childProcess?: ChildProcessWithoutNullStreams) {
  if (!childProcess?.pid) return false
  if (childProcess.exitCode !== null || childProcess.signalCode !== null) return false

  try {
    process.kill(childProcess.pid, 0)
    return true
  } catch {
    return false
  }
}

function stopSessionProcess (childProcess: ChildProcessWithoutNullStreams, streamKey: string) {
  const pid = childProcess.pid
  if (!pid || !isChildProcessAlive(childProcess)) return

  childProcess.kill('SIGTERM')

  const forceKillTimer = setTimeout(() => {
    if (!isChildProcessAlive(childProcess)) return

    logger.warn('Force killing stale live session ffmpeg process for %s.', streamKey, { pid })

    try {
      process.kill(pid, 'SIGKILL')
    } catch (err) {
      logger.warn('Cannot force kill live session ffmpeg process for %s.', streamKey, { err, pid })
    }
  }, 5_000)

  forceKillTimer.unref()
}

async function loadActiveScheduledLiveVideo (cameraId: string) {
  const now = new Date()
  const liveVideos = await LiveVideoModel.listForStarting({
    startBefore: now,
    endAfter: now
  })

  return liveVideos.find(liveVideo =>
    liveVideo.targetCamera === cameraId &&
    liveVideo.status === LiveVideoStatus.SCHEDULED
  )
}