// 视频转码任务封装，供上传接口和转码API复用
import { prisma } from '@/lib/prisma'
import { transcodeQueue } from '@/lib/video-transcode-queue'
import { processVideoTranscode, resolveLocalVideoPath } from '@/lib/video-processor'
import { access } from 'fs/promises'

export const USER_TRANSCODE_FAILURE_MESSAGE = '预览还没做好，请稍后重试'
import { deleteFileIfUnreferenced } from '@/lib/file-cleanup'

const MAX_TRANSCODE_RETRIES = 3

type CurrentLeaseResult<T> =
  | { current: true; value: T }
  | { current: false }

type LockedLeaseMaterial = {
  videoCoverUrl: string | null
  previewUrl: string
}

type TranscodeResult = Awaited<ReturnType<typeof processVideoTranscode>>

type CommittedTranscodeSnapshot = {
  videoProcessingStatus: 'done'
  videoPreview540Url: string
  videoPreviewClipUrl: string
  videoCoverUrl: string | null
  previewUrl: string
  videoDuration: number | null
  transcodeLastError: null
  transcodeAttempts: number
}

type ObservedTranscodeSnapshot = {
  videoProcessingStatus: string | null
  videoPreview540Url: string | null
  videoPreviewClipUrl: string | null
  videoCoverUrl: string | null
  previewUrl: string
  videoDuration: number | null
  transcodeLastError: string | null
  transcodeAttempts: number | null
}

type TranscodeFinalizeResult =
  | { state: 'committed' }
  | { state: 'not_committed'; error: unknown }
  | { state: 'unknown'; error: unknown; observed?: ObservedTranscodeSnapshot | null }

/**
 * 所有任务状态写入都以 PostgreSQL 的 clock_timestamp() 判定租约，不能使用
 * 应用容器的本地时间。FOR UPDATE 同时把“校验 token”和后续写入收在一个事务里。
 */
async function withCurrentDatabaseLease<T>(
  materialId: string,
  leaseToken: string,
  action: (tx: any, material: LockedLeaseMaterial) => Promise<T>,
): Promise<CurrentLeaseResult<T>> {
  return prisma.$transaction(async (tx) => {
    const rows = await tx.$queryRaw<Array<LockedLeaseMaterial>>`
      SELECT "videoCoverUrl", "previewUrl"
      FROM "materials"
      WHERE "id" = ${materialId}
        AND "transcode_lease_token" = ${leaseToken}
        AND "transcode_lease_expires_at" > clock_timestamp()
      FOR UPDATE
    `
    const material = rows[0]
    if (!material) {
      return { current: false } as const
    }
    return { current: true, value: await action(tx, material) } as const
  })
}

class NonRetryableTranscodeError extends Error {
  constructor(message: string) {
    super(message)
    this.name = 'NonRetryableTranscodeError'
  }
}

export class TranscodeCommitStateUnknownError extends Error {
  readonly originalError: unknown

  constructor(materialId: string, originalError: unknown) {
    super(`视频转码结果的数据库提交状态无法确认，已保留生成物等待恢复或人工对账: ${materialId}`)
    this.name = 'TranscodeCommitStateUnknownError'
    this.originalError = originalError
  }
}

function extractErrorMessage(error: unknown): string {
  if (error instanceof Error) {
    return error.message
  }

  if (typeof error === 'string') {
    return error
  }

  try {
    return JSON.stringify(error)
  } catch {
    return String(error)
  }
}

function isNonRetryableTranscodeError(error: unknown): boolean {
  if (error instanceof NonRetryableTranscodeError) {
    return true
  }

  const code = (error as { code?: string } | undefined)?.code
  if (code && code.toUpperCase() === 'ENOENT') {
    return true
  }

  const message = extractErrorMessage(error)
  return message.includes('FFmpeg not available')
}

function generatedOutputUrls(result: TranscodeResult): string[] {
  return Array.from(new Set([
    result.preview540Url,
    result.clipUrl,
    result.coverUrl,
  ].filter((url): url is string => Boolean(url?.trim()))))
}

/**
 * 任务拿到 processor 结果后，只有在明确未提交时才会调用这里。仍通过全库引用
 * 复核再删，防止租约切换期间另一个任务或用户操作已经引用了相同 URL。
 */
export async function cleanupUncommittedTranscodeOutputs(
  materialId: string,
  result: TranscodeResult,
): Promise<void> {
  const cleanupResults = await Promise.allSettled(
    generatedOutputUrls(result).map((url) => deleteFileIfUnreferenced(url)),
  )

  for (const cleanupResult of cleanupResults) {
    if (cleanupResult.status === 'rejected') {
      console.error(
        `[Transcode] Failed to clean an uncommitted output for material ${materialId}; manual reconciliation is required:`,
        cleanupResult.reason,
      )
    } else if (!cleanupResult.value.success) {
      console.error(
        `[Transcode] Uncommitted output cleanup did not succeed for material ${materialId}:`,
        cleanupResult.value.error,
      )
    }
  }
}

function snapshotMatches(
  observed: ObservedTranscodeSnapshot,
  expected: CommittedTranscodeSnapshot,
): boolean {
  return observed.videoProcessingStatus === expected.videoProcessingStatus
    && observed.videoPreview540Url === expected.videoPreview540Url
    && observed.videoPreviewClipUrl === expected.videoPreviewClipUrl
    && observed.videoCoverUrl === expected.videoCoverUrl
    && observed.previewUrl === expected.previewUrl
    && observed.videoDuration === expected.videoDuration
    && observed.transcodeLastError === expected.transcodeLastError
    && observed.transcodeAttempts === expected.transcodeAttempts
}

function observedReferencesGeneratedOutput(
  observed: ObservedTranscodeSnapshot,
  result: TranscodeResult,
): boolean {
  const generatedUrls = new Set(generatedOutputUrls(result))
  return [
    observed.videoPreview540Url,
    observed.videoPreviewClipUrl,
    observed.videoCoverUrl,
    observed.previewUrl,
  ].some((url) => Boolean(url && generatedUrls.has(url)))
}

async function readCommittedTranscodeSnapshot(
  materialId: string,
): Promise<ObservedTranscodeSnapshot | null> {
  return prisma.material.findUnique({
    where: { id: materialId },
    select: {
      videoProcessingStatus: true,
      videoPreview540Url: true,
      videoPreviewClipUrl: true,
      videoCoverUrl: true,
      previewUrl: true,
      videoDuration: true,
      transcodeLastError: true,
      transcodeAttempts: true,
    },
  })
}

async function finalizeTranscodeResult(
  materialId: string,
  leaseToken: string,
  attempt: number,
  result: TranscodeResult,
): Promise<TranscodeFinalizeResult> {
  let expectedSnapshot: CommittedTranscodeSnapshot | null = null

  try {
    const completed = await withCurrentDatabaseLease<{ count: number }>(materialId, leaseToken, async (tx, lockedMaterial) => {
      const currentCoverUrl = lockedMaterial.videoCoverUrl?.trim()
        ? lockedMaterial.videoCoverUrl
        : null
      const currentPreviewUrl = lockedMaterial.previewUrl?.trim()
        ? lockedMaterial.previewUrl
        : null
      const automaticCoverUrl = result.coverUrl || null

      expectedSnapshot = {
        videoProcessingStatus: 'done',
        videoPreview540Url: result.preview540Url!,
        videoPreviewClipUrl: result.clipUrl!,
        videoCoverUrl: currentCoverUrl || automaticCoverUrl,
        previewUrl: currentPreviewUrl || currentCoverUrl || automaticCoverUrl || '',
        videoDuration: result.duration,
        transcodeLastError: null,
        transcodeAttempts: attempt,
      }

      return tx.material.updateMany({
        where: { id: materialId, transcodeLeaseToken: leaseToken },
        data: {
          videoProcessingStatus: 'done',
          videoPreview540Url: result.preview540Url,
          videoPreviewClipUrl: result.clipUrl,
          // 只填空值；事务锁保证判断与写入之间不会被并发更新穿透。
          videoCoverUrl: currentCoverUrl ? undefined : automaticCoverUrl || undefined,
          videoDuration: result.duration,
          transcodeLastError: null,
          transcodeAttempts: attempt,
          previewUrl: currentPreviewUrl
            ? undefined
            : currentCoverUrl || automaticCoverUrl || undefined,
        },
      })
    })

    if (!completed.current || completed.value.count !== 1) {
      return {
        state: 'not_committed',
        error: new StaleTranscodeLeaseError(materialId),
      }
    }

    return { state: 'committed' }
  } catch (commitError) {
    let observed: ObservedTranscodeSnapshot | null
    try {
      observed = await readCommittedTranscodeSnapshot(materialId)
    } catch (verificationError) {
      console.error(
        `[Transcode] Cannot verify completion after database error for material ${materialId}; preserving outputs:`,
        verificationError,
      )
      return { state: 'unknown', error: commitError }
    }

    if (expectedSnapshot && observed && snapshotMatches(observed, expectedSnapshot)) {
      console.warn(
        `[Transcode] Completion acknowledgement was lost for material ${materialId}, but the exact committed state was verified`,
      )
      return { state: 'committed' }
    }

    if (observed && observedReferencesGeneratedOutput(observed, result)) {
      console.error(
        `[Transcode] Completion state is partial or changed for material ${materialId}; preserving outputs for reconciliation`,
      )
      return { state: 'unknown', error: commitError, observed }
    }

    // PostgreSQL 事务已返回失败，且主库回读确认没有任何本次唯一 URL 被引用。
    // 这是“明确未提交”；实际物理删除仍由 deleteFileIfUnreferenced 二次复核。
    return { state: 'not_committed', error: commitError }
  }
}

/**
 * 实际的转码任务处理函数
 * 
 * 内部包含最多 3 次重试逻辑，队列外部无需处理 retry。
 */
export async function executeTranscodeTask(
  materialId: string,
  fallbackVideoSourceUrl: string,
  needsCoverImage: boolean,
  leaseToken: string,
  abortSignal: AbortSignal = new AbortController().signal,
): Promise<void> {
  let attempt = 0
  const throwIfInterrupted = () => {
    if (abortSignal.aborted) {
      // 心跳、租约或 active marker 的状态无法继续证明时，当前执行结果必须
      // 视为陈旧。这个分支绝不消耗重试次数，也绝不把可恢复任务写成 failed。
      throw new StaleTranscodeLeaseError(materialId)
    }
  }

  while (attempt < MAX_TRANSCODE_RETRIES) {
    throwIfInterrupted()
    attempt += 1
    let attemptResult: TranscodeResult | null = null
    let outputDisposition: 'none' | 'pending' | 'cleaned' | 'preserve' | 'committed' = 'none'

    try {
      console.log(`[Transcode] Starting task for material ${materialId} (attempt ${attempt}/${MAX_TRANSCODE_RETRIES})`)

      const latestMaterial = await prisma.material.findUnique({
        where: { id: materialId },
        select: {
          videoSourceUrl: true,
          videoCoverUrl: true,
          previewUrl: true,
        },
      })

      const videoSourceUrl = latestMaterial?.videoSourceUrl || fallbackVideoSourceUrl
      if (!videoSourceUrl) {
        throw new NonRetryableTranscodeError(USER_TRANSCODE_FAILURE_MESSAGE)
      }

      // 入队时的 needsCoverImage 只是快照；真正开始处理时以数据库最新值为准，
      // 避免排队期间用户已补手动封面却仍被自动截图覆盖。
      const coverStillNeeded = !latestMaterial?.videoCoverUrl?.trim()
      if (needsCoverImage !== coverStillNeeded) {
        console.log(`[Transcode] Cover requirement changed while queued for material ${materialId}; using latest database value`)
      }

      const fileExists = await validateVideoFile(videoSourceUrl)
      if (!fileExists) {
        throw new Error(USER_TRANSCODE_FAILURE_MESSAGE)
      }

      const started = await withCurrentDatabaseLease<{ count: number }>(materialId, leaseToken, (tx) => tx.material.updateMany({
        where: { id: materialId, transcodeLeaseToken: leaseToken },
        data: { videoProcessingStatus: 'processing', transcodeLastError: null },
      }))
      if (!started.current || started.value.count !== 1) {
        throw new StaleTranscodeLeaseError(materialId)
      }

      attemptResult = await processVideoTranscode({
        inputPath: videoSourceUrl,
        outputDir: './uploads/transcoded',
        materialId,
        needsCoverImage: coverStillNeeded,
        abortSignal,
        onProgress: (progress, stage) => {
          console.log(`[Transcode] Material ${materialId} - ${stage}: ${progress}%`)
          void transcodeQueue.updateProgress(materialId, leaseToken, progress, stage)
            .catch((error) => console.error(`[Transcode] Failed to persist progress for ${materialId}:`, error))
        },
      })
      outputDisposition = 'pending'
      const result = attemptResult

      throwIfInterrupted()

      if (!result.preview540Url || !result.clipUrl || (coverStillNeeded && !result.coverUrl)) {
        throw new NonRetryableTranscodeError('视频转码输出不完整，拒绝标记为完成')
      }

      // 锁住当前素材后只填空封面/首页预览。若事务响应丢失，必须精确回读全部
      // 结果字段；只有 exact match 才当作成功，部分匹配一律保留对象等待对账。
      const finalized = await finalizeTranscodeResult(materialId, leaseToken, attempt, result)
      if (finalized.state === 'unknown') {
        outputDisposition = 'preserve'
        throw new TranscodeCommitStateUnknownError(materialId, finalized.error)
      }
      if (finalized.state === 'not_committed') {
        await cleanupUncommittedTranscodeOutputs(materialId, result)
        outputDisposition = 'cleaned'
        throw finalized.error
      }
      outputDisposition = 'committed'

      console.log(`[Transcode] Task completed for material ${materialId}`)
      return
    } catch (error) {
      if (attemptResult && outputDisposition === 'pending') {
        // processor 已成功返回，而完成事务尚未被确认；所有进入 unknown 的分支
        // 都会先把 disposition 改为 preserve，因此这里仅覆盖明确未提交阶段。
        await cleanupUncommittedTranscodeOutputs(materialId, attemptResult)
        outputDisposition = 'cleaned'
      }
      if (error instanceof TranscodeCommitStateUnknownError) {
        // 不能再写 failed 或重跑 FFmpeg；原事务可能已经提交。
        throw error
      }
      if (abortSignal.aborted) {
        throw new StaleTranscodeLeaseError(materialId)
      }
      if (error instanceof StaleTranscodeLeaseError) {
        // 蓝绿切换或租约过期后，旧进程不得把成功/失败结果覆盖到新持有者。
        throw error
      }
      const errorMsg = extractErrorMessage(error)
      console.error(`[Transcode] Task attempt ${attempt} failed for material ${materialId}:`, errorMsg)

      const shouldRetry = attempt < MAX_TRANSCODE_RETRIES && !isNonRetryableTranscodeError(error)

      if (!shouldRetry) {
        const failed = await withCurrentDatabaseLease<{ count: number }>(materialId, leaseToken, (tx) => tx.material.updateMany({
          where: { id: materialId, transcodeLeaseToken: leaseToken },
          data: {
            videoProcessingStatus: 'failed',
            transcodeLastError: USER_TRANSCODE_FAILURE_MESSAGE,
            transcodeAttempts: attempt,
          },
        }))
        if (!failed.current || failed.value.count !== 1) {
          throw new StaleTranscodeLeaseError(materialId)
        }
        const finalError = error instanceof Error ? error : new Error(errorMsg)
        throw finalError
      }

      const retryRecorded = await withCurrentDatabaseLease<{ count: number }>(materialId, leaseToken, (tx) => tx.material.updateMany({
        where: { id: materialId, transcodeLeaseToken: leaseToken },
        data: {
          transcodeLastError: USER_TRANSCODE_FAILURE_MESSAGE,
          transcodeAttempts: attempt,
        },
      }))
      if (!retryRecorded.current || retryRecorded.value.count !== 1) {
        throw new StaleTranscodeLeaseError(materialId)
      }

      console.warn(`[Transcode] Retrying material ${materialId} (${attempt}/${MAX_TRANSCODE_RETRIES})`)
    }
  }
}

export class StaleTranscodeLeaseError extends Error {
  constructor(materialId: string) {
    super(`视频转码租约已失效，拒绝写入旧进程结果: ${materialId}`)
    this.name = 'StaleTranscodeLeaseError'
  }
}

/**
 * 验证视频文件是否存在
 */
async function validateVideoFile(videoSourceUrl: string): Promise<boolean> {
  try {
    if (videoSourceUrl.startsWith('http://') || videoSourceUrl.startsWith('https://')) {
      return true
    }

    const localFilePath = resolveLocalVideoPath(videoSourceUrl)
    await access(localFilePath)
    console.log(`[TranscodeTask] Video source is ready: ${localFilePath}`)
    return true
  } catch (error) {
    console.error('[TranscodeTask] Video source is not readable', {
      videoSourceUrl,
      error,
    })
    return false
  }
}

/**
 * 入队一个视频转码任务
 */
export async function enqueueVideoTranscodeTask(
  materialId: string,
  videoSourceUrl: string,
  needsCoverImage: boolean
): Promise<void> {
  // 验证视频文件是否存在
  const fileExists = await validateVideoFile(videoSourceUrl)
  if (!fileExists) {
    throw new Error(USER_TRANSCODE_FAILURE_MESSAGE)
  }

  console.log(`[TranscodeTask] 📝 Enqueuing transcode task: materialId=${materialId}, videoSourceUrl=${videoSourceUrl}, needsCoverImage=${needsCoverImage}`)

  // 只等待数据库持久化入队，不等待 ffmpeg；失败需要反馈给调用方，不能静默丢任务。
  await transcodeQueue.add(materialId, videoSourceUrl, needsCoverImage, executeTranscodeTask)

  console.log(`[TranscodeTask] ✅ Transcode task enqueued successfully (fire-and-forget): materialId=${materialId}`)
}