/**
 * 视频转码持久化队列。
 *
 * 内存队列只负责降低当前进程的调度开销，数据库中的阶段记录和素材租约才是事实来源。
 * 每次执行前必须先取得全局唯一租约，因此蓝绿容器并存时也只会有一个 ffmpeg 任务。
 */

import { randomUUID } from 'crypto'
import { access, mkdir, rm, writeFile } from 'fs/promises'
import path from 'path'
import { LEGACY_COMPATIBLE_VIDEO_MATERIAL_TYPES } from './video-material-types'

export const VIDEO_MATERIAL_TYPES = [...LEGACY_COMPATIBLE_VIDEO_MATERIAL_TYPES]

export const VIDEO_TRANSCODE_DRAIN_PROTOCOL = 'VIDEO_TRANSCODE_DRAIN_PROTOCOL_V1'
export const DEFAULT_TRANSCODE_LEASE_MS = 5 * 60 * 1000
export const DEFAULT_TRANSCODE_HEARTBEAT_MS = 30 * 1000
export const LEGACY_TRANSCODE_STALE_MS = 10 * 60 * 1000

const SAFE_MAX_CONCURRENT_IN_SHARED_CONTAINER = 1
const GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY = 193582731
// 不直接 SELECT 返回 void 的 pg_advisory_xact_lock；Prisma 无法反序列化 void 列。
const ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL = 'SELECT 1 AS locked FROM pg_advisory_xact_lock($1)'
// 必须在取得 advisory lock 后读取数据库当前时钟，不能使用进入事务前的应用时间。
const READ_DATABASE_CLOCK_SQL = 'SELECT clock_timestamp() AS "now"'

export type TranscodeStage = 'preview' | 'clip' | 'cover'

export type TranscodeTaskFn = (
  materialId: string,
  videoSourceUrl: string,
  needsCoverImage: boolean,
  leaseToken: string,
  abortSignal: AbortSignal,
) => Promise<void>

interface QueuedTask {
  materialId: string
  videoSourceUrl: string
  needsCoverImage: boolean
  taskFn: TranscodeTaskFn
}

interface RecoverInterruptedTasksOptions {
  requeuePending?: boolean
  limit?: number
  videoTypes?: string[]
}

interface VideoTranscodeQueueOptions {
  maxConcurrent?: number
  prisma?: any
  workerId?: string
  instanceId?: string
  runtimeRoot?: string
  leaseDurationMs?: number
  heartbeatIntervalMs?: number
  drainAccess?: typeof access
}

type LeaseClaim =
  | { status: 'acquired'; token: string }
  | { status: 'busy' }
  | { status: 'skip' }

function safeWorkerId(value: string): string {
  return value.replace(/[^A-Za-z0-9_.-]/g, '_').slice(0, 120) || `worker-${process.pid}`
}

function uniqueTasks(tasks: Array<{ materialId: string; videoSourceUrl: string; needsCoverImage: boolean }>) {
  const seen = new Set<string>()
  return tasks.filter((task) => {
    if (!task.materialId || !task.videoSourceUrl || seen.has(task.materialId)) {
      return false
    }
    seen.add(task.materialId)
    return true
  })
}

async function readDatabaseClock(client: any): Promise<Date> {
  const rows = await client.$queryRawUnsafe(READ_DATABASE_CLOCK_SQL)
  const value = rows?.[0]?.now
  const now = value instanceof Date ? value : new Date(value)
  if (!value || Number.isNaN(now.getTime())) {
    throw new Error('无法读取数据库当前时间，拒绝执行视频转码租约操作')
  }
  return now
}

export class VideoTranscodeQueue {
  private queue: QueuedTask[] = []
  private processing = new Set<string>()
  private progress = new Map<string, number>()
  private stages = new Map<string, TranscodeStage>()
  private prisma: any
  private readonly maxConcurrent: number
  private readonly workerId: string
  private readonly instanceId: string
  private readonly runtimeRoot: string
  private readonly leaseDurationMs: number
  private readonly heartbeatIntervalMs: number
  private readonly drainAccess: typeof access
  private pumping = false
  private recoveryPromise: Promise<number> | null = null
  private retryTimer: ReturnType<typeof setTimeout> | null = null

  constructor(options: VideoTranscodeQueueOptions | number = {}) {
    const normalizedOptions = typeof options === 'number' ? { maxConcurrent: options } : options
    const envConcurrent = Number.parseInt(process.env.VIDEO_TRANSCODE_MAX_CONCURRENT || '', 10)
    const requested = Number.isFinite(envConcurrent)
      ? envConcurrent
      : normalizedOptions.maxConcurrent ?? SAFE_MAX_CONCURRENT_IN_SHARED_CONTAINER

    this.maxConcurrent = Math.max(1, Math.min(SAFE_MAX_CONCURRENT_IN_SHARED_CONTAINER, requested))
    this.prisma = normalizedOptions.prisma || null
    this.workerId = safeWorkerId(
      normalizedOptions.workerId || process.env.VIDEO_TRANSCODE_WORKER_ID || process.env.HOSTNAME || `worker-${process.pid}`,
    )
    this.instanceId = safeWorkerId(
      normalizedOptions.instanceId
      || process.env.VIDEO_TRANSCODE_INSTANCE_ID
      || `${this.workerId}-${process.pid}`,
    )
    this.runtimeRoot = normalizedOptions.runtimeRoot
      || process.env.VIDEO_TRANSCODE_RUNTIME_DIR
      || path.join(process.env.UPLOAD_DIR || path.join(process.cwd(), 'uploads'), '.video-transcode-runtime')
    this.leaseDurationMs = normalizedOptions.leaseDurationMs || DEFAULT_TRANSCODE_LEASE_MS
    this.heartbeatIntervalMs = normalizedOptions.heartbeatIntervalMs || DEFAULT_TRANSCODE_HEARTBEAT_MS
    this.drainAccess = normalizedOptions.drainAccess || access

    console.log(`[TranscodeQueue] Initialized worker=${this.workerId}, maxConcurrent=${this.maxConcurrent}`)
  }

  private async getPrisma() {
    if (!this.prisma) {
      const { prisma } = await import('@/lib/prisma')
      this.prisma = prisma
    }
    return this.prisma
  }

  private capabilityPath() {
    return path.join(this.runtimeRoot, 'capabilities', `${this.workerId}.v1`)
  }

  private drainPath() {
    return path.join(this.runtimeRoot, 'drain', this.workerId)
  }

  private activePath() {
    return path.join(this.runtimeRoot, 'active', `${this.workerId}.json`)
  }

  async ensureDrainProtocolCapability(): Promise<void> {
    const target = this.capabilityPath()
    await mkdir(path.dirname(target), { recursive: true })
    await writeFile(
      target,
      `${VIDEO_TRANSCODE_DRAIN_PROTOCOL}\nworkerId=${this.workerId}\ninstanceId=${this.instanceId}\n`,
      { encoding: 'utf8', mode: 0o600 },
    )
  }

  private async isDraining(): Promise<boolean> {
    try {
      await this.drainAccess(this.drainPath())
      return true
    } catch (error) {
      const code = (error as NodeJS.ErrnoException | undefined)?.code
      if (code === 'ENOENT') {
        return false
      }

      // NAS/runtime 状态无法读取时必须 fail-closed。把 EACCES/EIO/ESTALE
      // 等错误视为仍在 drain，pump 会保留任务并在下一轮重试，绝不启动 FFmpeg。
      console.error(`[TranscodeQueue] Cannot verify drain marker for ${this.workerId}; treating worker as draining:`, error)
      return true
    }
  }

  private async writeActiveMarker(materialId: string, leaseToken: string): Promise<void> {
    const target = this.activePath()
    await mkdir(path.dirname(target), { recursive: true })
    await writeFile(target, JSON.stringify({
      protocol: VIDEO_TRANSCODE_DRAIN_PROTOCOL,
      workerId: this.workerId,
      instanceId: this.instanceId,
      materialId,
      leaseToken,
      pid: process.pid,
      heartbeatAt: new Date().toISOString(),
    }), { encoding: 'utf8', mode: 0o600 })
  }

  private async removeActiveMarker(): Promise<void> {
    await rm(this.activePath(), { force: true }).catch((error) => {
      console.error(`[TranscodeQueue] Failed to remove active marker for ${this.workerId}:`, error)
    })
  }

  private async saveStageProgress(
    materialId: string,
    stage: TranscodeStage,
    status: 'waiting' | 'processing' | 'completed' | 'failed',
    progress = 0,
    errorMessage: string | null = null,
    leaseToken?: string,
  ): Promise<boolean> {
    const prisma = await this.getPrisma()
    const writeProgress = async (client: any) => {
      await client.videoTranscodeProgress.upsert({
        where: { materialId_stage: { materialId, stage } },
        create: {
          materialId,
          stage,
          status,
          progress,
          errorMessage,
          lastHeartbeat: new Date(),
        },
        update: {
          status,
          progress,
          errorMessage,
          lastHeartbeat: status === 'processing' ? new Date() : undefined,
          completedAt: status === 'completed' || status === 'failed' ? new Date() : undefined,
        },
      })
    }

    if (!leaseToken) {
      await writeProgress(prisma)
      return true
    }

    // 进度表本身没有租约列，因此必须和 claim/release 共用同一把事务锁，
    // 在锁内校验 token 后再写入，避免旧进程在校验与 upsert 之间越过新租约。
    return prisma.$transaction(async (tx: any) => {
      await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
      const databaseNow = await readDatabaseClock(tx)
      const current = await tx.material.count({
        where: {
          id: materialId,
          transcodeLeaseToken: leaseToken,
          transcodeLeaseExpiresAt: { gt: databaseNow },
        },
      })
      if (current !== 1) {
        return false
      }
      await writeProgress(tx)
      return true
    })
  }

  private async persistWaitingStages(task: Pick<QueuedTask, 'materialId' | 'needsCoverImage'>): Promise<void> {
    const stages: TranscodeStage[] = ['preview', 'clip']
    if (task.needsCoverImage) {
      stages.push('cover')
    }
    for (const stage of stages) {
      await this.saveStageProgress(task.materialId, stage, 'waiting')
    }
  }

  private async prepareMaterialForQueue(materialId: string): Promise<boolean> {
    const prisma = await this.getPrisma()
    const prepared = await prisma.$transaction(async (tx: any) => {
      await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
      const databaseNow = await readDatabaseClock(tx)
      return tx.material.updateMany({
        where: {
          id: materialId,
          deletedAt: null,
          videoSourceUrl: { not: null },
          OR: [
            { transcodeLeaseToken: null },
            { transcodeLeaseExpiresAt: null },
            { transcodeLeaseExpiresAt: { lte: databaseNow } },
          ],
        },
        data: { videoProcessingStatus: 'pending' },
      })
    })
    return prepared.count === 1
  }

  private enqueueLocal(task: QueuedTask): boolean {
    if (this.processing.has(task.materialId) || this.queue.some((queued) => queued.materialId === task.materialId)) {
      return false
    }
    this.queue.push(task)
    return true
  }

  /** 持久化等待状态后立即返回，不等待 ffmpeg 完成。 */
  async add(
    materialId: string,
    videoSourceUrl: string,
    needsCoverImage: boolean,
    taskFn: TranscodeTaskFn,
  ): Promise<void> {
    const task = { materialId, videoSourceUrl, needsCoverImage, taskFn }
    if (this.processing.has(materialId) || this.queue.some((queued) => queued.materialId === materialId)) {
      return
    }

    if (!(await this.prepareMaterialForQueue(materialId))) {
      return
    }
    await this.persistWaitingStages(task)
    if (this.enqueueLocal(task)) {
      void this.pump()
    }
  }

  private schedulePump(delayMs = 2_000): void {
    if (this.retryTimer) {
      return
    }
    this.retryTimer = setTimeout(() => {
      this.retryTimer = null
      void this.pump()
    }, delayMs)
    this.retryTimer.unref?.()
  }

  private async tryAcquireGlobalLease(materialId: string): Promise<LeaseClaim> {
    const prisma = await this.getPrisma()
    const token = randomUUID()

    return prisma.$transaction(async (tx: any) => {
      await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
      const databaseNow = await readDatabaseClock(tx)
      const expiresAt = new Date(databaseNow.getTime() + this.leaseDurationMs)

      const activeLease = await tx.material.findFirst({
        where: {
          transcodeLeaseToken: { not: null },
          transcodeLeaseExpiresAt: { gt: databaseNow },
        },
        select: { id: true },
      })
      if (activeLease) {
        return { status: 'busy' } as const
      }

      const material = await tx.material.findUnique({
        where: { id: materialId },
        select: {
          id: true,
          deletedAt: true,
          videoSourceUrl: true,
          videoProcessingStatus: true,
        },
      })
      if (!material || material.deletedAt || !material.videoSourceUrl || material.videoProcessingStatus === 'done') {
        return { status: 'skip' } as const
      }

      const claimed = await tx.material.updateMany({
        where: {
          id: materialId,
          OR: [
            { transcodeLeaseToken: null },
            { transcodeLeaseExpiresAt: null },
            { transcodeLeaseExpiresAt: { lte: databaseNow } },
          ],
        },
        data: {
          transcodeLeaseToken: token,
          transcodeLeaseExpiresAt: expiresAt,
          videoProcessingStatus: 'processing',
          transcodeLastError: null,
        },
      })

      return claimed.count === 1
        ? { status: 'acquired', token } as const
        : { status: 'busy' } as const
    })
  }

  async isLeaseCurrent(materialId: string, leaseToken: string): Promise<boolean> {
    const prisma = await this.getPrisma()
    return prisma.$transaction(async (tx: any) => {
      await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
      const databaseNow = await readDatabaseClock(tx)
      const count = await tx.material.count({
        where: {
          id: materialId,
          transcodeLeaseToken: leaseToken,
          transcodeLeaseExpiresAt: { gt: databaseNow },
        },
      })
      return count === 1
    })
  }

  async renewLease(materialId: string, leaseToken: string): Promise<boolean> {
    const prisma = await this.getPrisma()
    return prisma.$transaction(async (tx: any) => {
      // renew/acquire/release/progress 必须共用同一把事务锁。这样旧 worker
      // 的续租不可能在新 worker 已领取另一素材后再“复活”为第二个有效租约。
      await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
      const databaseNow = await readDatabaseClock(tx)
      const renewed = await tx.material.updateMany({
        where: {
          id: materialId,
          transcodeLeaseToken: leaseToken,
          transcodeLeaseExpiresAt: { gt: databaseNow },
        },
        data: { transcodeLeaseExpiresAt: new Date(databaseNow.getTime() + this.leaseDurationMs) },
      })
      return renewed.count === 1
    })
  }

  async releaseLease(materialId: string, leaseToken: string, resetToPending = false): Promise<boolean> {
    const prisma = await this.getPrisma()
    return prisma.$transaction(async (tx: any) => {
      await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
      const released = await tx.material.updateMany({
        where: { id: materialId, transcodeLeaseToken: leaseToken },
        data: {
          transcodeLeaseToken: null,
          transcodeLeaseExpiresAt: null,
          ...(resetToPending ? { videoProcessingStatus: 'pending' } : {}),
        },
      })
      return released.count === 1
    })
  }

  private async runClaimedTask(task: QueuedTask, leaseToken: string): Promise<void> {
    this.processing.add(task.materialId)
    let leaseLost = false
    let heartbeatTimer: ReturnType<typeof setTimeout> | null = null
    let heartbeatInFlight: Promise<void> = Promise.resolve()
    let heartbeatStopped = false
    let requeueAfterDrain = false
    const abortController = new AbortController()
    const scheduleHeartbeat = () => {
      heartbeatTimer = setTimeout(() => {
        heartbeatTimer = null
        heartbeatInFlight = (async () => {
          try {
            const renewed = await this.renewLease(task.materialId, leaseToken)
            if (!renewed) {
              leaseLost = true
              heartbeatStopped = true
              abortController.abort(new Error('视频转码租约已失效'))
              console.error(`[TranscodeQueue] Lease lost for ${task.materialId}; stale result writes will be rejected`)
              return
            }
            await this.writeActiveMarker(task.materialId, leaseToken)
          } catch (error) {
            leaseLost = true
            heartbeatStopped = true
            abortController.abort(error instanceof Error ? error : new Error(String(error)))
            console.error(`[TranscodeQueue] Lease heartbeat failed for ${task.materialId}:`, error)
          }
        })().finally(() => {
          if (!heartbeatStopped) {
            scheduleHeartbeat()
          }
        })
      }, this.heartbeatIntervalMs)
      ;(heartbeatTimer as NodeJS.Timeout).unref?.()
    }

    try {
      // active 标记写入成功后才允许启动 ffmpeg；标记失败时保持 fail-closed。
      await this.writeActiveMarker(task.materialId, leaseToken)
      // drain 可能恰好出现在 pump 二次检查与 active 标记写入之间。
      // 写入 active 后再核对一次；命中时由 finally 释放租约、清理标记并重新入队，绝不启动 ffmpeg。
      if (await this.isDraining()) {
        requeueAfterDrain = true
        console.log(`[TranscodeQueue] Drain detected after active marker for ${task.materialId}; task stays pending`)
        return
      }
      scheduleHeartbeat()

      await task.taskFn(
        task.materialId,
        task.videoSourceUrl,
        task.needsCoverImage,
        leaseToken,
        abortController.signal,
      )
      if (leaseLost || !(await this.isLeaseCurrent(task.materialId, leaseToken))) {
        console.warn(`[TranscodeQueue] Ignoring completion from stale worker for ${task.materialId}`)
      } else {
        console.log(`[TranscodeQueue] Task completed for ${task.materialId}`)
      }
    } catch (error) {
      console.error(`[TranscodeQueue] Task failed for ${task.materialId}:`, error)
    } finally {
      heartbeatStopped = true
      if (heartbeatTimer) {
        clearTimeout(heartbeatTimer)
      }
      // 等待已经开始的续租/active 标记写入结束，再释放租约并删除标记，
      // 防止异步心跳在 finally 之后重新制造一个陈旧 active 文件。
      await heartbeatInFlight.catch(() => {})
      this.progress.delete(task.materialId)
      this.stages.delete(task.materialId)
      // drain 或心跳/marker/租约中断都属于可恢复中断。只有当前 token 仍归本
      // worker 时才会清租约并回 pending；若 token 已被新 worker 接管，CAS 无操作。
      const resetInterruptedTask = requeueAfterDrain || leaseLost || abortController.signal.aborted
      await this.releaseLease(task.materialId, leaseToken, resetInterruptedTask).catch((error) => {
        console.error(`[TranscodeQueue] Lease release failed for ${task.materialId}:`, error)
      })
      await this.removeActiveMarker()
      this.processing.delete(task.materialId)
      if (requeueAfterDrain && !this.queue.some((queued) => queued.materialId === task.materialId)) {
        this.queue.unshift(task)
      }
      void this.pump()
    }
  }

  private async pump(): Promise<void> {
    if (this.pumping || this.processing.size >= this.maxConcurrent || this.queue.length === 0) {
      return
    }
    this.pumping = true
    try {
      await this.ensureDrainProtocolCapability()
      if (await this.isDraining()) {
        this.schedulePump(2_000)
        return
      }

      const task = this.queue.shift()
      if (!task) {
        return
      }
      const claim = await this.tryAcquireGlobalLease(task.materialId)
      if (claim.status === 'busy') {
        this.queue.push(task)
        this.schedulePump(2_000)
        return
      }
      if (claim.status === 'skip') {
        this.progress.delete(task.materialId)
        this.stages.delete(task.materialId)
        this.schedulePump(0)
        return
      }

      // drain 标记可能在数据库 claim 进行期间出现；取得租约后再次检查，
      // 确保部署脚本确认 idle 后不会再启动新的 ffmpeg。
      if (await this.isDraining()) {
        await this.releaseLease(task.materialId, claim.token, true)
        this.queue.unshift(task)
        this.schedulePump(2_000)
        return
      }

      void this.runClaimedTask(task, claim.token)
    } catch (error) {
      console.error('[TranscodeQueue] Queue pump failed:', error)
      this.schedulePump(2_000)
    } finally {
      this.pumping = false
    }
  }

  async updateProgress(
    materialId: string,
    leaseToken: string,
    progress: number,
    stage?: TranscodeStage,
  ): Promise<boolean> {
    const normalized = Math.min(100, Math.max(0, progress))
    if (stage) {
      const saved = await this.saveStageProgress(
        materialId,
        stage,
        normalized >= 100 ? 'completed' : 'processing',
        normalized,
        null,
        leaseToken,
      )
      if (!saved) {
        return false
      }
      this.progress.set(materialId, normalized)
      this.stages.set(materialId, stage)
      return true
    }
    if (!(await this.isLeaseCurrent(materialId, leaseToken))) {
      return false
    }
    this.progress.set(materialId, normalized)
    return true
  }

  getStatus() {
    return {
      queueLength: this.queue.length,
      processingCount: this.processing.size,
      processingIds: Array.from(this.processing),
      maxConcurrent: this.maxConcurrent,
      progress: this.progress,
      stages: this.stages,
      waiting: this.queue.map((task) => task.materialId),
    }
  }

  getDetailedStatus() {
    return {
      queue: this.queue.map((task) => task.materialId),
      processing: Array.from(this.processing),
      maxConcurrent: this.maxConcurrent,
    }
  }

  async loadWaitingTasks(limit?: number): Promise<Array<{ materialId: string; videoSourceUrl: string; needsCoverImage: boolean }>> {
    const prisma = await this.getPrisma()
    const rows = await prisma.videoTranscodeProgress.findMany({
      where: { status: { in: ['waiting', 'processing'] } },
      include: {
        material: {
          select: {
            videoSourceUrl: true,
            videoCoverUrl: true,
            videoProcessingStatus: true,
            deletedAt: true,
          },
        },
      },
      orderBy: [{ priority: 'desc' }, { createdAt: 'asc' }],
      ...(limit && limit > 0 ? { take: limit * 3 } : {}),
    })

    const tasks = rows
      .filter((row: any) => row.material
        && !row.material.deletedAt
        && row.material.videoSourceUrl
        && row.material.videoProcessingStatus !== 'done'
        && row.material.videoProcessingStatus !== 'failed')
      .map((row: any) => ({
        materialId: String(row.materialId),
        videoSourceUrl: String(row.material.videoSourceUrl),
        needsCoverImage: !row.material.videoCoverUrl,
      }))
    return uniqueTasks(tasks).slice(0, limit && limit > 0 ? limit : undefined)
  }

  private async recoverExpiredAndLegacyLeases(videoTypes: string[]): Promise<number> {
    const prisma = await this.getPrisma()
    // 预筛也必须使用数据库时钟。最终回收仍会在 advisory lock 内再次以
    // 数据库时钟做 CAS，避免应用容器时钟偏快/偏慢造成漏恢复或误恢复。
    const databaseNow = await readDatabaseClock(prisma)
    const staleBefore = new Date(databaseNow.getTime() - LEGACY_TRANSCODE_STALE_MS)
    let recovered = 0

    const expired = await prisma.material.findMany({
      where: {
        materialType: { in: videoTypes },
        videoProcessingStatus: 'processing',
        videoSourceUrl: { not: null },
        deletedAt: null,
        transcodeLeaseToken: { not: null },
        transcodeLeaseExpiresAt: { lte: databaseNow },
      },
      select: { id: true, transcodeLeaseToken: true, transcodeLeaseExpiresAt: true },
    })

    for (const material of expired) {
      const released = await prisma.$transaction(async (tx: any) => {
        await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
        const databaseNow = await readDatabaseClock(tx)
        const result = await tx.material.updateMany({
          where: {
            id: material.id,
            transcodeLeaseToken: material.transcodeLeaseToken,
            transcodeLeaseExpiresAt: { lte: databaseNow },
          },
          data: {
            videoProcessingStatus: 'pending',
            transcodeLeaseToken: null,
            transcodeLeaseExpiresAt: null,
          },
        })
        return result.count
      })
      recovered += released
    }

    // 兼容首次上线前留下的旧任务：只有素材和阶段心跳都已静默 10 分钟才回收。
    const legacy = await prisma.material.findMany({
      where: {
        materialType: { in: videoTypes },
        videoProcessingStatus: 'processing',
        videoSourceUrl: { not: null },
        deletedAt: null,
        transcodeLeaseToken: null,
        updatedAt: { lt: staleBefore },
      },
      select: {
        id: true,
        updatedAt: true,
        transcodeProgresses: {
          select: { lastHeartbeat: true },
          orderBy: { lastHeartbeat: 'desc' },
          take: 1,
        },
      },
    })

    for (const material of legacy) {
      const heartbeat = material.transcodeProgresses?.[0]?.lastHeartbeat
      if (heartbeat && heartbeat >= staleBefore) {
        continue
      }
      const released = await prisma.$transaction(async (tx: any) => {
        await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
        const result = await tx.material.updateMany({
          where: {
            id: material.id,
            videoProcessingStatus: 'processing',
            transcodeLeaseToken: null,
            updatedAt: material.updatedAt,
          },
          data: { videoProcessingStatus: 'pending' },
        })
        return result.count
      })
      recovered += released
    }

    return recovered
  }

  private async loadPendingMaterials(videoTypes: string[], limit?: number) {
    const prisma = await this.getPrisma()
    const rows = await prisma.material.findMany({
      where: {
        materialType: { in: videoTypes },
        videoProcessingStatus: 'pending',
        videoSourceUrl: { not: null },
        deletedAt: null,
      },
      orderBy: { createdAt: 'asc' },
      select: { id: true, videoSourceUrl: true, videoCoverUrl: true },
      ...(limit && limit > 0 ? { take: limit } : {}),
    })
    return rows.map((row: any) => ({
      materialId: String(row.id),
      videoSourceUrl: String(row.videoSourceUrl),
      needsCoverImage: !row.videoCoverUrl,
    }))
  }

  /**
   * 回收过期租约并重新装载持久化任务。恢复过程不删除任何 S3 对象或 canonical 源文件。
   * 同一进程的重复扫描会复用同一个 Promise，避免 60 秒定时器重叠。
   */
  async runRecoveryScan(
    taskFn: TranscodeTaskFn,
    options: RecoverInterruptedTasksOptions = {},
  ): Promise<number> {
    if (this.recoveryPromise) {
      return this.recoveryPromise
    }

    this.recoveryPromise = (async () => {
      await this.ensureDrainProtocolCapability()
      const videoTypes = options.videoTypes?.length ? options.videoTypes : VIDEO_MATERIAL_TYPES
      const recoveredLeases = await this.recoverExpiredAndLegacyLeases(videoTypes)
      const persisted = await this.loadWaitingTasks(options.limit)
      const pending = options.requeuePending === false
        ? []
        : await this.loadPendingMaterials(videoTypes, options.limit)
      const tasks = uniqueTasks([...persisted, ...pending])
      let enqueued = 0

      for (const task of tasks) {
        if (this.enqueueLocal({ ...task, taskFn })) {
          enqueued += 1
        }
      }
      if (enqueued > 0) {
        void this.pump()
      }
      console.log(`[TranscodeQueue] Recovery scan: released=${recoveredLeases}, enqueued=${enqueued}`)
      return enqueued
    })()

    try {
      return await this.recoveryPromise
    } finally {
      this.recoveryPromise = null
    }
  }

  async recoverOnStartup(taskFn: TranscodeTaskFn): Promise<number> {
    return this.runRecoveryScan(taskFn, { requeuePending: true, videoTypes: VIDEO_MATERIAL_TYPES })
  }

  /** 兼容旧调用签名；cleanupFn 被有意忽略，恢复不得删除当前引用对象。 */
  async recoverInterruptedTasks(
    taskFn: TranscodeTaskFn,
    _cleanupFn?: unknown,
    options: RecoverInterruptedTasksOptions = {},
  ): Promise<{ recovered: number; requeued: number }> {
    const requeued = await this.runRecoveryScan(taskFn, { ...options, requeuePending: true })
    return { recovered: 0, requeued }
  }
}

export const transcodeQueue = new VideoTranscodeQueue()