/** @jest-environment node */

import {
  executeTranscodeTask,
  StaleTranscodeLeaseError,
  TranscodeCommitStateUnknownError,
} from '@/lib/video-transcode-task'
import { prisma } from '@/lib/prisma'
import { processVideoTranscode } from '@/lib/video-processor'
import { deleteFileIfUnreferenced } from '@/lib/file-cleanup'

jest.mock('@/lib/prisma', () => ({
  prisma: {
    material: {
      findUnique: jest.fn(),
      updateMany: jest.fn(),
    },
    $queryRaw: jest.fn(),
    $transaction: jest.fn(),
  },
}))

jest.mock('@/lib/video-processor', () => ({
  processVideoTranscode: jest.fn(),
}))

jest.mock('@/lib/file-cleanup', () => ({
  deleteFileIfUnreferenced: jest.fn(),
}))

jest.mock('@/lib/video-transcode-queue', () => ({
  transcodeQueue: {
    updateProgress: jest.fn().mockResolvedValue(true),
  },
}))

describe('executeTranscodeTask lease fencing', () => {
  beforeEach(() => {
    jest.clearAllMocks()
    ;(prisma.material.findUnique as jest.Mock).mockResolvedValue({
      videoSourceUrl: 'https://s3.jiansun.vip/source.mp4',
      videoCoverUrl: null,
      previewUrl: '',
    })
    ;(prisma.$queryRaw as jest.Mock).mockResolvedValue([{
      videoCoverUrl: null,
      previewUrl: '',
    }])
    ;(prisma.$transaction as jest.Mock).mockImplementation(
      async (callback: (tx: typeof prisma) => Promise<unknown>) => callback(prisma),
    )
    ;(processVideoTranscode as jest.Mock).mockResolvedValue({
      preview540Url: 'https://s3.jiansun.vip/preview.mp4',
      clipUrl: 'https://s3.jiansun.vip/clip.mp4',
      coverUrl: 'https://s3.jiansun.vip/cover.jpg',
      duration: 12,
    })
    ;(deleteFileIfUnreferenced as jest.Mock).mockResolvedValue({ success: true, skipped: false })
  })

  test('租约在完成前失效时，旧 token 不能写入完成结果或失败状态', async () => {
    ;(prisma.material.updateMany as jest.Mock)
      .mockResolvedValueOnce({ count: 1 })
      .mockResolvedValueOnce({ count: 0 })

    await expect(executeTranscodeTask(
      'material-1',
      'https://s3.jiansun.vip/source.mp4',
      true,
      'stale-token',
    )).rejects.toBeInstanceOf(StaleTranscodeLeaseError)

    expect(prisma.material.updateMany).toHaveBeenCalledTimes(2)
    expect(prisma.material.updateMany).not.toHaveBeenCalledWith(expect.objectContaining({
      data: expect.objectContaining({ videoProcessingStatus: 'failed' }),
    }))
    expect(deleteFileIfUnreferenced).toHaveBeenCalledTimes(3)
    expect(deleteFileIfUnreferenced).toHaveBeenCalledWith('https://s3.jiansun.vip/preview.mp4')
    expect(deleteFileIfUnreferenced).toHaveBeenCalledWith('https://s3.jiansun.vip/clip.mp4')
    expect(deleteFileIfUnreferenced).toHaveBeenCalledWith('https://s3.jiansun.vip/cover.jpg')
  })

  test('开始写入就发现 token 失效时立即停止，不运行 ffmpeg', async () => {
    ;(prisma.material.updateMany as jest.Mock).mockResolvedValueOnce({ count: 0 })

    await expect(executeTranscodeTask(
      'material-2',
      'https://s3.jiansun.vip/source.mp4',
      true,
      'expired-token',
    )).rejects.toBeInstanceOf(StaleTranscodeLeaseError)

    expect(processVideoTranscode).not.toHaveBeenCalled()
  })

  test('完成事务响应丢失但精确回读一致时视为已提交并保留生成物', async () => {
    const commitAckLost = new Error('connection closed after commit')
    let transactionCall = 0
    ;(prisma.$transaction as jest.Mock).mockImplementation(
      async (callback: (tx: typeof prisma) => Promise<unknown>) => {
        transactionCall += 1
        const value = await callback(prisma)
        if (transactionCall === 2) throw commitAckLost
        return value
      },
    )
    ;(prisma.material.findUnique as jest.Mock)
      .mockResolvedValueOnce({
        videoSourceUrl: 'https://s3.jiansun.vip/source.mp4',
        videoCoverUrl: null,
        previewUrl: '',
      })
      .mockResolvedValueOnce({
        videoProcessingStatus: 'done',
        videoPreview540Url: 'https://s3.jiansun.vip/preview.mp4',
        videoPreviewClipUrl: 'https://s3.jiansun.vip/clip.mp4',
        videoCoverUrl: 'https://s3.jiansun.vip/cover.jpg',
        previewUrl: 'https://s3.jiansun.vip/cover.jpg',
        videoDuration: 12,
        transcodeLastError: null,
        transcodeAttempts: 1,
      })
    ;(prisma.material.updateMany as jest.Mock)
      .mockResolvedValueOnce({ count: 1 })
      .mockResolvedValueOnce({ count: 1 })

    await expect(executeTranscodeTask(
      'material-commit-ack-lost',
      'https://s3.jiansun.vip/source.mp4',
      true,
      'valid-token',
    )).resolves.toBeUndefined()

    expect(prisma.material.findUnique).toHaveBeenCalledTimes(2)
    expect(deleteFileIfUnreferenced).not.toHaveBeenCalled()
    expect(prisma.material.updateMany).not.toHaveBeenCalledWith(expect.objectContaining({
      data: expect.objectContaining({ videoProcessingStatus: 'failed' }),
    }))
  })

  test('完成事务报错且回读只有部分结果时保留生成物并抛出可对账错误', async () => {
    let transactionCall = 0
    ;(prisma.$transaction as jest.Mock).mockImplementation(
      async (callback: (tx: typeof prisma) => Promise<unknown>) => {
        transactionCall += 1
        const value = await callback(prisma)
        if (transactionCall === 2) throw new Error('commit response unavailable')
        return value
      },
    )
    ;(prisma.material.findUnique as jest.Mock)
      .mockResolvedValueOnce({
        videoSourceUrl: 'https://s3.jiansun.vip/source.mp4',
        videoCoverUrl: null,
        previewUrl: '',
      })
      .mockResolvedValueOnce({
        videoProcessingStatus: 'processing',
        videoPreview540Url: 'https://s3.jiansun.vip/preview.mp4',
        videoPreviewClipUrl: null,
        videoCoverUrl: null,
        previewUrl: '',
        videoDuration: null,
        transcodeLastError: null,
        transcodeAttempts: 1,
      })
    ;(prisma.material.updateMany as jest.Mock)
      .mockResolvedValueOnce({ count: 1 })
      .mockResolvedValueOnce({ count: 1 })

    await expect(executeTranscodeTask(
      'material-partial-commit-state',
      'https://s3.jiansun.vip/source.mp4',
      true,
      'valid-token',
    )).rejects.toBeInstanceOf(TranscodeCommitStateUnknownError)

    expect(deleteFileIfUnreferenced).not.toHaveBeenCalled()
    expect(prisma.material.updateMany).not.toHaveBeenCalledWith(expect.objectContaining({
      data: expect.objectContaining({ videoProcessingStatus: 'failed' }),
    }))
  })

  test('完成事务报错且主库回读也失败时保留生成物，不猜测为未提交', async () => {
    let transactionCall = 0
    ;(prisma.$transaction as jest.Mock).mockImplementation(
      async (callback: (tx: typeof prisma) => Promise<unknown>) => {
        transactionCall += 1
        const value = await callback(prisma)
        if (transactionCall === 2) throw new Error('commit response unavailable')
        return value
      },
    )
    ;(prisma.material.findUnique as jest.Mock)
      .mockResolvedValueOnce({
        videoSourceUrl: 'https://s3.jiansun.vip/source.mp4',
        videoCoverUrl: null,
        previewUrl: '',
      })
      .mockRejectedValueOnce(new Error('database read unavailable'))
    ;(prisma.material.updateMany as jest.Mock)
      .mockResolvedValueOnce({ count: 1 })
      .mockResolvedValueOnce({ count: 1 })

    await expect(executeTranscodeTask(
      'material-unknown-commit-state',
      'https://s3.jiansun.vip/source.mp4',
      true,
      'valid-token',
    )).rejects.toBeInstanceOf(TranscodeCommitStateUnknownError)

    expect(deleteFileIfUnreferenced).not.toHaveBeenCalled()
    expect(prisma.material.updateMany).not.toHaveBeenCalledWith(expect.objectContaining({
      data: expect.objectContaining({ videoProcessingStatus: 'failed' }),
    }))
  })

  test('FFmpeg 没有生成完整预览时不能把素材误标为 done', async () => {
    ;(prisma.material.updateMany as jest.Mock)
      .mockResolvedValueOnce({ count: 1 })
      .mockResolvedValueOnce({ count: 1 })
    ;(processVideoTranscode as jest.Mock).mockResolvedValueOnce({
      preview540Url: null,
      clipUrl: null,
      coverUrl: null,
      duration: null,
    })

    await expect(executeTranscodeTask(
      'material-3',
      'https://s3.jiansun.vip/source.mp4',
      true,
      'valid-token',
    )).rejects.toThrow('视频转码输出不完整')

    expect(prisma.material.updateMany).toHaveBeenLastCalledWith(expect.objectContaining({
      data: expect.objectContaining({ videoProcessingStatus: 'failed' }),
    }))
    expect(prisma.material.updateMany).not.toHaveBeenCalledWith(expect.objectContaining({
      data: expect.objectContaining({ videoProcessingStatus: 'done' }),
    }))
  })

  test('已有用户封面时跳过重新截图，但首页 previewUrl 仍沿用该封面', async () => {
    ;(prisma.material.findUnique as jest.Mock).mockResolvedValueOnce({
      videoSourceUrl: 'https://s3.jiansun.vip/source.mp4',
      videoCoverUrl: 'https://s3.jiansun.vip/user-cover.jpg',
      previewUrl: '',
    })
    ;(prisma.material.updateMany as jest.Mock)
      .mockResolvedValueOnce({ count: 1 })
      .mockResolvedValueOnce({ count: 1 })
    ;(prisma.$queryRaw as jest.Mock).mockResolvedValue([{
      videoCoverUrl: 'https://s3.jiansun.vip/user-cover.jpg',
      previewUrl: '',
    }])
    ;(processVideoTranscode as jest.Mock).mockResolvedValueOnce({
      preview540Url: 'https://s3.jiansun.vip/preview.mp4',
      clipUrl: 'https://s3.jiansun.vip/clip.mp4',
      coverUrl: null,
      duration: 12,
    })

    await expect(executeTranscodeTask(
      'material-4',
      'https://s3.jiansun.vip/source.mp4',
      false,
      'valid-token',
    )).resolves.toBeUndefined()

    expect(processVideoTranscode).toHaveBeenCalledWith(expect.objectContaining({ needsCoverImage: false }))
    expect(prisma.material.updateMany).toHaveBeenLastCalledWith(expect.objectContaining({
      data: expect.objectContaining({
        videoProcessingStatus: 'done',
        previewUrl: 'https://s3.jiansun.vip/user-cover.jpg',
      }),
    }))
  })

  test('转码期间用户补写封面和首页预览时，完成事务只写转码结果且不覆盖用户内容', async () => {
    let finishProcessing!: (value: {
      preview540Url: string
      clipUrl: string
      coverUrl: string
      duration: number
    }) => void
    let markProcessingStarted!: () => void
    const processingStarted = new Promise<void>((resolve) => {
      markProcessingStarted = resolve
    })
    const processingResult = new Promise((resolve) => {
      finishProcessing = resolve
    })
    ;(processVideoTranscode as jest.Mock).mockImplementationOnce(() => {
      markProcessingStarted()
      return processingResult
    })
    ;(prisma.material.updateMany as jest.Mock)
      .mockResolvedValueOnce({ count: 1 })
      .mockResolvedValueOnce({ count: 1 })

    const taskPromise = executeTranscodeTask(
      'material-concurrent-user-media',
      'https://s3.jiansun.vip/source.mp4',
      true,
      'valid-token',
    )

    await processingStarted
    expect(processVideoTranscode).toHaveBeenCalledTimes(1)

    // 模拟 FFmpeg 运行期间用户手动补写了两项内容；完成事务必须读取这份最新值。
    ;(prisma.$queryRaw as jest.Mock).mockResolvedValueOnce([{
      videoCoverUrl: 'https://s3.jiansun.vip/manual-cover-during-transcode.jpg',
      previewUrl: 'https://s3.jiansun.vip/manual-preview-during-transcode.jpg',
    }])
    finishProcessing({
      preview540Url: 'https://s3.jiansun.vip/preview.mp4',
      clipUrl: 'https://s3.jiansun.vip/clip.mp4',
      coverUrl: 'https://s3.jiansun.vip/automatic-cover.jpg',
      duration: 12,
    })

    await expect(taskPromise).resolves.toBeUndefined()

    expect(prisma.$queryRaw).toHaveBeenCalledTimes(2)
    expect(prisma.material.updateMany).toHaveBeenLastCalledWith(expect.objectContaining({
      where: expect.objectContaining({
        transcodeLeaseToken: 'valid-token',
      }),
      data: expect.objectContaining({
        videoProcessingStatus: 'done',
        videoPreview540Url: 'https://s3.jiansun.vip/preview.mp4',
        videoPreviewClipUrl: 'https://s3.jiansun.vip/clip.mp4',
        videoDuration: 12,
      }),
    }))
    const completionData = (prisma.material.updateMany as jest.Mock).mock.calls.at(-1)?.[0]?.data
    expect(completionData.videoCoverUrl).toBeUndefined()
    expect(completionData.previewUrl).toBeUndefined()
  })

  test('排队期间用户已补手动封面时，以开始处理时最新值为准并跳过自动截图', async () => {
    ;(prisma.material.findUnique as jest.Mock).mockResolvedValueOnce({
      videoSourceUrl: 'https://s3.jiansun.vip/source.mp4',
      videoCoverUrl: 'https://s3.jiansun.vip/manual-cover-before-start.jpg',
      previewUrl: '',
    })
    ;(prisma.$queryRaw as jest.Mock).mockResolvedValue([{
      videoCoverUrl: 'https://s3.jiansun.vip/manual-cover-before-start.jpg',
      previewUrl: '',
    }])
    ;(prisma.material.updateMany as jest.Mock)
      .mockResolvedValueOnce({ count: 1 })
      .mockResolvedValueOnce({ count: 1 })
    ;(processVideoTranscode as jest.Mock).mockResolvedValueOnce({
      preview540Url: 'https://s3.jiansun.vip/preview.mp4',
      clipUrl: 'https://s3.jiansun.vip/clip.mp4',
      coverUrl: null,
      duration: 12,
    })

    await expect(executeTranscodeTask(
      'material-cover-added-while-queued',
      'https://s3.jiansun.vip/source.mp4',
      true,
      'valid-token',
    )).resolves.toBeUndefined()

    expect(processVideoTranscode).toHaveBeenCalledWith(expect.objectContaining({ needsCoverImage: false }))
    const completionData = (prisma.material.updateMany as jest.Mock).mock.calls.at(-1)?.[0]?.data
    expect(completionData.videoCoverUrl).toBeUndefined()
    expect(completionData.previewUrl).toBe('https://s3.jiansun.vip/manual-cover-before-start.jpg')
  })

  test('应用时钟即使快十年，任务状态写入仍只使用数据库时钟租约门禁', async () => {
    jest.useFakeTimers().setSystemTime(new Date('2036-08-11T00:00:00Z'))
    try {
      ;(prisma.material.updateMany as jest.Mock)
        .mockResolvedValueOnce({ count: 1 })
        .mockResolvedValueOnce({ count: 1 })

      await expect(executeTranscodeTask(
        'material-clock-skew',
        'https://s3.jiansun.vip/source.mp4',
        true,
        'database-valid-token',
      )).resolves.toBeUndefined()

      const leaseQueries = (prisma.$queryRaw as jest.Mock).mock.calls.map((call) => (
        Array.from(call[0] as TemplateStringsArray).join(' ')
      ))
      expect(leaseQueries).toHaveLength(2)
      expect(leaseQueries.every((sql) => sql.includes('clock_timestamp()'))).toBe(true)
      for (const [args] of (prisma.material.updateMany as jest.Mock).mock.calls) {
        expect(args.where).not.toHaveProperty('transcodeLeaseExpiresAt')
      }
    } finally {
      jest.useRealTimers()
    }
  })

  test('心跳中断信号不会重试或把仍有源文件的任务写成 failed', async () => {
    const controller = new AbortController()
    let beginProcessing!: () => void
    const processingStarted = new Promise<void>((resolve) => { beginProcessing = resolve })
    ;(prisma.material.updateMany as jest.Mock).mockResolvedValueOnce({ count: 1 })
    ;(processVideoTranscode as jest.Mock).mockImplementationOnce(async ({ abortSignal }) => {
      beginProcessing()
      await new Promise<void>((resolve) => abortSignal.addEventListener('abort', () => resolve(), { once: true }))
      throw new Error('ffmpeg aborted after heartbeat failure')
    })

    const task = executeTranscodeTask(
      'material-heartbeat-interrupted',
      'https://s3.jiansun.vip/source.mp4',
      true,
      'still-owned-token',
      controller.signal,
    )
    await processingStarted
    controller.abort(new Error('heartbeat unavailable'))

    await expect(task).rejects.toBeInstanceOf(StaleTranscodeLeaseError)
    expect(processVideoTranscode).toHaveBeenCalledTimes(1)
    expect(prisma.material.updateMany).not.toHaveBeenCalledWith(expect.objectContaining({
      data: expect.objectContaining({ videoProcessingStatus: 'failed' }),
    }))
    expect(prisma.material.updateMany).not.toHaveBeenCalledWith(expect.objectContaining({
      data: expect.objectContaining({ transcodeAttempts: 2 }),
    }))
  })
})