/** @jest-environment node */

import { access, mkdtemp, mkdir, readFile, rm, stat, writeFile } from 'fs/promises'
import { tmpdir } from 'os'
import path from 'path'
import { VideoTranscodeQueue } from '@/lib/video-transcode-queue'

type MaterialState = {
  id: string
  materialType: string
  videoSourceUrl: string | null
  videoCoverUrl: string | null
  videoPreview540Url?: string | null
  videoProcessingStatus: string | null
  transcodeLeaseToken: string | null
  transcodeLeaseExpiresAt: Date | null
  transcodeLastError?: string | null
  deletedAt: Date | null
  createdAt: Date
  updatedAt: Date
}

function createPrismaHarness(
  initial: MaterialState[],
  databaseClock: () => Date = () => new Date(),
) {
  const materials = new Map(initial.map((item) => [item.id, { ...item }]))
  const progress = new Map<string, any>()
  let transactionTail = Promise.resolve()

  const matches = (material: MaterialState, where: any): boolean => {
    if (!where) return true
    if (where.id !== undefined && material.id !== where.id) return false
    if (typeof where.videoProcessingStatus === 'string' && material.videoProcessingStatus !== where.videoProcessingStatus) return false
    if (where.videoProcessingStatus?.not !== undefined && material.videoProcessingStatus === where.videoProcessingStatus.not) return false
    if (where.deletedAt === null && material.deletedAt !== null) return false
    if (where.materialType?.in && !where.materialType.in.includes(material.materialType)) return false
    if (where.videoSourceUrl?.not === null && material.videoSourceUrl === null) return false
    if (where.transcodeLeaseToken === null && material.transcodeLeaseToken !== null) return false
    if (typeof where.transcodeLeaseToken === 'string' && material.transcodeLeaseToken !== where.transcodeLeaseToken) return false
    if (where.transcodeLeaseToken?.not === null && material.transcodeLeaseToken === null) return false
    if (where.transcodeLeaseExpiresAt === null && material.transcodeLeaseExpiresAt !== null) return false
    if (where.transcodeLeaseExpiresAt?.gt && (!material.transcodeLeaseExpiresAt || material.transcodeLeaseExpiresAt <= where.transcodeLeaseExpiresAt.gt)) return false
    if (where.transcodeLeaseExpiresAt?.lte && (!material.transcodeLeaseExpiresAt || material.transcodeLeaseExpiresAt > where.transcodeLeaseExpiresAt.lte)) return false
    if (where.updatedAt instanceof Date && material.updatedAt.getTime() !== where.updatedAt.getTime()) return false
    if (where.OR && !where.OR.some((part: any) => matches(material, part))) return false
    return true
  }

  const prisma: any = {
    $queryRawUnsafe: jest.fn(async (sql: string) => {
      if (sql.includes('clock_timestamp()')) {
        return [{ now: databaseClock() }]
      }
      return []
    }),
    $transaction: jest.fn((callback: (tx: any) => Promise<any>) => {
      let release!: () => void
      const previous = transactionTail
      transactionTail = new Promise<void>((resolve) => { release = resolve })
      return previous.then(() => callback(prisma)).finally(release)
    }),
    material: {
      findFirst: jest.fn(async ({ where }: any) => Array.from(materials.values()).find((item) => matches(item, where)) || null),
      findUnique: jest.fn(async ({ where }: any) => materials.get(where.id) || null),
      findMany: jest.fn(async ({ where }: any) => Array.from(materials.values())
        .filter((item) => matches(item, where))
        .map((item) => ({ ...item, transcodeProgresses: [] }))),
      count: jest.fn(async ({ where }: any) => Array.from(materials.values()).filter((item) => matches(item, where)).length),
      updateMany: jest.fn(async ({ where, data }: any) => {
        let count = 0
        for (const item of Array.from(materials.values())) {
          if (matches(item, where)) {
            Object.assign(item, data, { updatedAt: new Date() })
            count += 1
          }
        }
        return { count }
      }),
    },
    videoTranscodeProgress: {
      upsert: jest.fn(async ({ where, create, update }: any) => {
        const key = `${where.materialId_stage.materialId}:${where.materialId_stage.stage}`
        const current = progress.get(key)
        const next = current ? { ...current, ...update } : { ...create }
        progress.set(key, next)
        return next
      }),
      findMany: jest.fn(async () => Array.from(progress.entries()).map(([key, row]) => {
        const materialId = key.split(':')[0]
        return { ...row, materialId, material: materials.get(materialId) }
      })),
    },
  }

  return { prisma, materials, progress }
}

function material(id: string): MaterialState {
  return {
    id,
    materialType: '淘系-主图视频',
    videoSourceUrl: `/uploads/${id}.mp4`,
    videoCoverUrl: null,
    videoProcessingStatus: 'pending',
    transcodeLeaseToken: null,
    transcodeLeaseExpiresAt: null,
    deletedAt: null,
    createdAt: new Date('2026-08-11T00:00:00Z'),
    updatedAt: new Date('2026-08-11T00:00:00Z'),
  }
}

async function waitUntil(predicate: () => boolean, timeoutMs = 5_000) {
  const deadline = Date.now() + timeoutMs
  while (!predicate()) {
    if (Date.now() > deadline) throw new Error('waitUntil timeout')
    await new Promise((resolve) => setTimeout(resolve, 10))
  }
}

function isQueueIdle(queue: VideoTranscodeQueue): boolean {
  const status = queue.getStatus()
  return status.queueLength === 0 && status.processingCount === 0
}

async function wakeAndDrainQueueForTest(queue: VideoTranscodeQueue): Promise<void> {
  await waitUntil(() => !(queue as any).pumping, 2_000)
  const retryTimer = (queue as any).retryTimer as ReturnType<typeof setTimeout> | null
  if (retryTimer) {
    clearTimeout(retryTimer)
    ;(queue as any).retryTimer = null
  }
  await (queue as any).pump()
  try {
    await waitUntil(() => isQueueIdle(queue), 5_000)
  } catch {
    throw new Error(`queue did not become idle: ${JSON.stringify({
      status: queue.getDetailedStatus(),
      pumping: (queue as any).pumping,
      retryScheduled: Boolean((queue as any).retryTimer),
    })}`)
  }
}

async function stopDrainedQueueForTest(queue: VideoTranscodeQueue): Promise<void> {
  await waitUntil(() => !(queue as any).pumping, 2_000)
  const retryTimer = (queue as any).retryTimer as ReturnType<typeof setTimeout> | null
  if (retryTimer) {
    clearTimeout(retryTimer)
    ;(queue as any).retryTimer = null
  }
  ;((queue as any).queue as unknown[]).length = 0
  await waitUntil(() => isQueueIdle(queue))
}

describe('video transcode database lease', () => {
  let runtimeRoot: string

  beforeEach(async () => {
    runtimeRoot = await mkdtemp(path.join(tmpdir(), 'video-lease-test-'))
  })

  afterEach(async () => {
    await rm(runtimeRoot, { recursive: true, force: true })
  })

  test('add 在任务完成前返回，转码继续在后台执行', async () => {
    const harness = createPrismaHarness([material('a')])
    const queue = new VideoTranscodeQueue({
      prisma: harness.prisma,
      workerId: 'worker-a',
      instanceId: 'worker-a-instance-1',
      runtimeRoot,
      heartbeatIntervalMs: 20,
    })
    let finish!: () => void
    const taskFinished = new Promise<void>((resolve) => { finish = resolve })
    let started = false

    await queue.add('a', '/uploads/a.mp4', true, async () => {
      started = true
      await taskFinished
    })

    await waitUntil(() => started)
    expect(queue.getStatus().processingCount).toBe(1)
    const capability = await readFile(path.join(runtimeRoot, 'capabilities', 'worker-a.v1'), 'utf8')
    expect(capability).toContain('workerId=worker-a\n')
    expect(capability).toContain('instanceId=worker-a-instance-1\n')
    const activeMarker = path.join(runtimeRoot, 'active', 'worker-a.json')
    const firstHeartbeat = (await stat(activeMarker)).mtimeMs
    await new Promise((resolve) => setTimeout(resolve, 60))
    expect((await stat(activeMarker)).mtimeMs).toBeGreaterThan(firstHeartbeat)
    finish()
    await waitUntil(() => isQueueIdle(queue))
    await expect(access(activeMarker)).rejects.toBeDefined()
  })

  test('两个容器共享数据库时，全局最多只有一个有效租约', async () => {
    const harness = createPrismaHarness([material('a'), material('b')])
    const queueA = new VideoTranscodeQueue({ prisma: harness.prisma, workerId: 'blue', runtimeRoot })
    const queueB = new VideoTranscodeQueue({ prisma: harness.prisma, workerId: 'green', runtimeRoot })
    const releases = new Map<string, () => void>()
    const starts: string[] = []
    let active = 0
    let maxActive = 0
    const task = async (id: string) => {
      starts.push(id)
      active += 1
      maxActive = Math.max(maxActive, active)
      await new Promise<void>((resolve) => releases.set(id, resolve))
      active -= 1
    }

    await Promise.all([
      queueA.add('a', '/uploads/a.mp4', true, task),
      queueB.add('b', '/uploads/b.mp4', true, task),
    ])

    await waitUntil(() => starts.length === 1)
    expect(maxActive).toBe(1)
    releases.get(starts[0])?.()
    await waitUntil(() => starts.length === 2, 6_000)
    expect(maxActive).toBe(1)
    releases.get(starts[1])?.()
    await waitUntil(() => active === 0 && isQueueIdle(queueA) && isQueueIdle(queueB))
  }, 10_000)

  test('续租与新领取共用 advisory lock，等待中的旧续租不会在新租约后复活', async () => {
    const renewing = material('renewing')
    renewing.videoProcessingStatus = 'processing'
    renewing.transcodeLeaseToken = 'renew-token'
    renewing.transcodeLeaseExpiresAt = new Date(Date.now() + 30_000)
    const harness = createPrismaHarness([renewing, material('next-material')])
    const queue = new VideoTranscodeQueue({
      prisma: harness.prisma,
      workerId: 'renew-acquire-race',
      runtimeRoot,
      leaseDurationMs: 60_000,
    })

    const originalUpdateMany = harness.prisma.material.updateMany.getMockImplementation()!
    const originalFindFirst = harness.prisma.material.findFirst.getMockImplementation()!
    let allowRenewUpdate!: () => void
    const renewUpdateGate = new Promise<void>((resolve) => { allowRenewUpdate = resolve })
    let markRenewUpdateStarted!: () => void
    const renewUpdateStarted = new Promise<void>((resolve) => { markRenewUpdateStarted = resolve })
    const events: string[] = []

    harness.prisma.material.updateMany.mockImplementation(async (args: any) => {
      if (args.where?.id === 'renewing' && args.where?.transcodeLeaseToken === 'renew-token') {
        events.push('renew-update-start')
        markRenewUpdateStarted()
        await renewUpdateGate
        events.push('renew-update-finish')
      }
      return originalUpdateMany(args)
    })
    harness.prisma.material.findFirst.mockImplementation(async (args: any) => {
      events.push('acquire-active-scan')
      return originalFindFirst(args)
    })

    const renewPromise = queue.renewLease('renewing', 'renew-token')
    await renewUpdateStarted
    const acquirePromise = (queue as any).tryAcquireGlobalLease('next-material')

    // 第二个事务已提交给 Prisma，但 advisory lock 仍由续租事务持有，
    // 所以领取逻辑尚不能扫描或写入另一份租约。
    await Promise.resolve()
    await Promise.resolve()
    expect(events).toEqual(['renew-update-start'])

    allowRenewUpdate()
    await expect(renewPromise).resolves.toBe(true)
    await expect(acquirePromise).resolves.toEqual({ status: 'busy' })

    expect(events).toEqual([
      'renew-update-start',
      'renew-update-finish',
      'acquire-active-scan',
    ])
    expect(harness.materials.get('renewing')?.transcodeLeaseToken).toBe('renew-token')
    expect(harness.materials.get('next-material')?.transcodeLeaseToken).toBeNull()
    const leaseSql = harness.prisma.$queryRawUnsafe.mock.calls.map(([sql]: [string]) => sql)
    expect(leaseSql).toEqual([
      'SELECT 1 AS locked FROM pg_advisory_xact_lock($1)',
      'SELECT clock_timestamp() AS "now"',
      'SELECT 1 AS locked FROM pg_advisory_xact_lock($1)',
      'SELECT clock_timestamp() AS "now"',
    ])
  })

  test('仅回收已过期租约，保留 canonical 源文件和现有预览引用', async () => {
    // 数据库比应用机快很多：旧实现用应用时钟预筛会把这条租约误当成未来有效，
    // 新实现必须仍按数据库 clock_timestamp() 正确回收。
    const databaseNow = new Date('2099-08-12T00:00:00Z')
    const expired = material('expired')
    expired.videoProcessingStatus = 'processing'
    expired.transcodeLeaseToken = 'old-token'
    expired.transcodeLeaseExpiresAt = new Date(databaseNow.getTime() - 60_000)
    expired.videoPreview540Url = 'https://s3.jiansun.vip/existing-preview.mp4'
    const harness = createPrismaHarness([expired], () => databaseNow)
    harness.progress.set('expired:preview', { status: 'waiting', stage: 'preview', priority: 0, createdAt: new Date() })
    const queue = new VideoTranscodeQueue({ prisma: harness.prisma, workerId: 'recovery', runtimeRoot })
    await mkdir(path.join(runtimeRoot, 'drain'), { recursive: true })
    await writeFile(path.join(runtimeRoot, 'drain', 'recovery'), '')

    const recovered = await queue.runRecoveryScan(jest.fn(), { requeuePending: true })
    const after = harness.materials.get('expired')!

    expect(recovered).toBe(1)
    expect(after.transcodeLeaseToken).toBeNull()
    expect(after.transcodeLeaseExpiresAt).toBeNull()
    expect(after.videoProcessingStatus).toBe('pending')
    expect(after.videoSourceUrl).toBe('/uploads/expired.mp4')
    expect(after.videoPreview540Url).toBe('https://s3.jiansun.vip/existing-preview.mp4')

    // 解除测试 drain，让已重新入队的后台任务真正排空；不能把 retry timer 留到用例结束后。
    await rm(path.join(runtimeRoot, 'drain', 'recovery'), { force: true })
    await wakeAndDrainQueueForTest(queue)
  })

  test('应用时钟快于数据库时，不会把数据库仍有效的租约重置为 pending', async () => {
    const databaseNow = new Date('2020-08-12T00:00:00Z')
    const active = material('database-active')
    active.videoProcessingStatus = 'processing'
    active.transcodeLeaseToken = 'database-valid-token'
    active.transcodeLeaseExpiresAt = new Date(databaseNow.getTime() + 60_000)
    const harness = createPrismaHarness([active], () => databaseNow)
    const queue = new VideoTranscodeQueue({
      prisma: harness.prisma,
      workerId: 'clock-skew-worker',
      runtimeRoot,
    })
    const taskFn = jest.fn(async () => {})

    await queue.add('database-active', '/uploads/database-active.mp4', true, taskFn)

    expect(taskFn).not.toHaveBeenCalled()
    expect(queue.getStatus()).toMatchObject({ queueLength: 0, processingCount: 0 })
    expect(harness.materials.get('database-active')).toMatchObject({
      videoProcessingStatus: 'processing',
      transcodeLeaseToken: 'database-valid-token',
      transcodeLeaseExpiresAt: active.transcodeLeaseExpiresAt,
    })
  })

  test('续租失败会向正在运行的 ffmpeg 链路发送取消信号', async () => {
    const harness = createPrismaHarness([material('abort-me')])
    const queue = new VideoTranscodeQueue({
      prisma: harness.prisma,
      workerId: 'abort-worker',
      runtimeRoot,
      heartbeatIntervalMs: 20,
    })
    let started = false
    let aborted = false

    await queue.add('abort-me', '/uploads/abort-me.mp4', true, async (_id, _url, _cover, _token, signal) => {
      started = true
      await new Promise<void>((resolve) => {
        signal.addEventListener('abort', () => {
          aborted = true
          resolve()
        }, { once: true })
      })
    })

    await waitUntil(() => started)
    harness.materials.get('abort-me')!.transcodeLeaseToken = 'taken-by-another-worker'
    await waitUntil(() => aborted)
    await waitUntil(() => isQueueIdle(queue))
  })

  test('心跳瞬时异常会终止执行并把当前 token 的任务恢复为 pending', async () => {
    const harness = createPrismaHarness([material('heartbeat-error')])
    const queue = new VideoTranscodeQueue({
      prisma: harness.prisma,
      workerId: 'heartbeat-error-worker',
      runtimeRoot,
      heartbeatIntervalMs: 20,
    })
    const renewSpy = jest.spyOn(queue, 'renewLease').mockRejectedValueOnce(new Error('database heartbeat unavailable'))
    let started = false
    let aborted = false

    await queue.add('heartbeat-error', '/uploads/heartbeat-error.mp4', true, async (_id, _url, _cover, _token, signal) => {
      started = true
      await new Promise<void>((resolve) => signal.addEventListener('abort', () => {
        aborted = true
        resolve()
      }, { once: true }))
    })

    await waitUntil(() => started)
    await waitUntil(() => aborted)
    await waitUntil(() => isQueueIdle(queue))

    expect(renewSpy).toHaveBeenCalled()
    expect(harness.materials.get('heartbeat-error')).toMatchObject({
      videoProcessingStatus: 'pending',
      transcodeLeaseToken: null,
      transcodeLeaseExpiresAt: null,
      videoSourceUrl: '/uploads/heartbeat-error.mp4',
    })
    expect(harness.prisma.material.updateMany).not.toHaveBeenCalledWith(expect.objectContaining({
      data: expect.objectContaining({ videoProcessingStatus: 'failed' }),
    }))

    const recoveredTask = jest.fn(async () => {})
    await expect(queue.runRecoveryScan(recoveredTask)).resolves.toBeGreaterThanOrEqual(1)
    await waitUntil(() => isQueueIdle(queue))
    expect(recoveredTask).toHaveBeenCalled()
  })

  test('active 标记写入时才出现 drain，也不会调用 taskFn', async () => {
    const harness = createPrismaHarness([material('drain-race')])
    const queue = new VideoTranscodeQueue({
      prisma: harness.prisma,
      workerId: 'drain-race-worker',
      runtimeRoot,
      heartbeatIntervalMs: 20,
    })
    const taskFn = jest.fn(async () => {})
    const originalWriteActiveMarker = (queue as any).writeActiveMarker.bind(queue)
    let injectDrain = true
    ;(queue as any).writeActiveMarker = async (materialId: string, leaseToken: string) => {
      await originalWriteActiveMarker(materialId, leaseToken)
      if (injectDrain) {
        injectDrain = false
        await mkdir(path.join(runtimeRoot, 'drain'), { recursive: true })
        await writeFile(path.join(runtimeRoot, 'drain', 'drain-race-worker'), '')
      }
    }

    await queue.add('drain-race', '/uploads/drain-race.mp4', true, taskFn)
    await waitUntil(() => queue.getStatus().queueLength === 1 && queue.getStatus().processingCount === 0)

    expect(taskFn).not.toHaveBeenCalled()
    expect(harness.materials.get('drain-race')).toMatchObject({
      videoProcessingStatus: 'pending',
      transcodeLeaseToken: null,
      transcodeLeaseExpiresAt: null,
    })
    await expect(access(path.join(runtimeRoot, 'active', 'drain-race-worker.json'))).rejects.toBeDefined()

    // 任务在 drain 期间应保持排队；测试结束前显式取消本地测试队列和 retry timer，
    // 避免后台日志越过 Jest 用例边界。数据库中的 pending 事实仍保留给下次恢复扫描。
    await stopDrainedQueueForTest(queue)
    expect(taskFn).not.toHaveBeenCalled()
  })

  test.each(['EACCES', 'EIO', 'ESTALE'])(
    'drain 标记读取返回 %s 时 fail-closed，不领取或启动任务',
    async (errorCode) => {
      const harness = createPrismaHarness([material(`drain-${errorCode}`)])
      const drainAccess = jest.fn(async () => {
        throw Object.assign(new Error(`simulated ${errorCode}`), { code: errorCode })
      })
      const queue = new VideoTranscodeQueue({
        prisma: harness.prisma,
        workerId: `drain-${errorCode}-worker`,
        runtimeRoot,
        drainAccess,
      })
      const taskFn = jest.fn(async () => {})

      await queue.add(`drain-${errorCode}`, `/uploads/drain-${errorCode}.mp4`, true, taskFn)
      await waitUntil(() => !(queue as any).pumping)

      expect(drainAccess).toHaveBeenCalled()
      expect(taskFn).not.toHaveBeenCalled()
      expect(queue.getStatus()).toMatchObject({ queueLength: 1, processingCount: 0 })
      expect(harness.materials.get(`drain-${errorCode}`)).toMatchObject({
        videoProcessingStatus: 'pending',
        transcodeLeaseToken: null,
      })
      await stopDrainedQueueForTest(queue)
    },
  )

  test('drain 标记读取仅 ENOENT 表示未排空，允许正常领取并完成任务', async () => {
    const harness = createPrismaHarness([material('drain-enoent')])
    const drainAccess = jest.fn(async () => {
      throw Object.assign(new Error('simulated ENOENT'), { code: 'ENOENT' })
    })
    const queue = new VideoTranscodeQueue({
      prisma: harness.prisma,
      workerId: 'drain-enoent-worker',
      runtimeRoot,
      drainAccess,
    })
    const taskFn = jest.fn(async () => {})

    await queue.add('drain-enoent', '/uploads/drain-enoent.mp4', true, taskFn)
    await waitUntil(() => isQueueIdle(queue))

    expect(drainAccess).toHaveBeenCalled()
    expect(taskFn).toHaveBeenCalledTimes(1)
    expect(harness.materials.get('drain-enoent')).toMatchObject({
      transcodeLeaseToken: null,
    })
  })

  test('租约释放后旧 token 的阶段进度写入也会被拒绝', async () => {
    const leased = material('stale-progress')
    leased.videoProcessingStatus = 'processing'
    leased.transcodeLeaseToken = 'old-token'
    leased.transcodeLeaseExpiresAt = new Date(Date.now() + 60_000)
    const harness = createPrismaHarness([leased])
    const queue = new VideoTranscodeQueue({ prisma: harness.prisma, workerId: 'stale-worker', runtimeRoot })

    await expect(queue.releaseLease('stale-progress', 'old-token')).resolves.toBe(true)
    await expect(queue.updateProgress('stale-progress', 'old-token', 100, 'preview')).resolves.toBe(false)
    expect(harness.progress.size).toBe(0)
    expect(harness.prisma.$queryRawUnsafe).toHaveBeenCalledWith(
      'SELECT 1 AS locked FROM pg_advisory_xact_lock($1)',
      expect.any(Number),
    )
    expect(harness.prisma.$queryRawUnsafe.mock.calls.every(
      ([sql]: [string]) => !sql.startsWith('SELECT pg_advisory_xact_lock'),
    )).toBe(true)
  })
})