const JOBS = 'vk_jobs'
const BATCH = 20
const LEASE_MS = 60_000
const MAX_ATTEMPTS = 5
const BACKOFF_BASE_MS = 30_000

/** 到点后直接映射为状态机 worker 动作的 job。 */
const TASK_ACTION_KINDS = Object.freeze(['open_expire', 'hold_timeout', 'checkpoint_miss', 'contest_close'])
const WARN_KINDS = Object.freeze(['checkpoint_warn', 'review_warn'])

/**
 * vk_jobs 的领取与执行：`FOR UPDATE SKIP LOCKED` + 租约。每个 job 执行时重新读任务当前状态，
 * 状态已变就无事可做，所以重复或过时的 job 无害。
 *
 * @param {{
 *   db: import('knex').Knex,
 *   tasks: {
 *     actAsWorker(ctx: object, input: { task_id: string, action: any }): Promise<{ ok: boolean }>,
 *     warn(ctx: object, input: { task_id: string, kind: any }): Promise<void>,
 *     expireTransfers(ctx: object, input: { task_id: string }): Promise<void>,
 *   },
 *   logger: { error(message: string, fields: object): void },
 * }} deps
 */
export function createJobs({ db, tasks, logger }) {
  if (typeof db !== 'function') throw new Error('createJobs: db is required')

  /**
   * 领取并执行一批到期 job；返回本轮执行的条数。
   * @param {object} ctx
   * @returns {Promise<number>}
   */
  async function runDue(ctx) {
    const now = new Date()
    const jobs = await db.transaction(async (trx) => {
      const due = await trx(JOBS)
        .where((q) => q.where({ status: 'pending' }).where('run_at', '<=', now).orWhere((r) => r.where({ status: 'running' }).where('lease_until', '<', now)))
        .orderBy('run_at', 'asc')
        .limit(BATCH)
        .forUpdate()
        .skipLocked()
      if (due.length > 0) await trx(JOBS).whereIn('id', due.map((job) => job.id)).update({ status: 'running', lease_until: new Date(now.getTime() + LEASE_MS), updated_at: now })
      return due
    })
    for (const job of jobs) {
      // worker 的全局错误处理：单个 job 失败只记录并退避重试，不拖垮整轮。
      try {
        await execute(ctx, job)
        await db(JOBS).where({ id: job.id }).update({ status: 'done', lease_until: null, updated_at: new Date() })
      } catch (err) {
        const attempts = job.attempts + 1
        const message = err instanceof Error ? (err.stack || err.message) : String(err)
        logger.error('job 执行失败', { job_id: job.id, kind: job.kind, ref_id: job.ref_id, attempts, error: message })
        await db(JOBS).where({ id: job.id }).update({
          status: attempts >= MAX_ATTEMPTS ? 'failed' : 'pending',
          attempts,
          run_at: new Date(Date.now() + BACKOFF_BASE_MS * 2 ** job.attempts),
          lease_until: null,
          last_error: message,
          updated_at: new Date(),
        })
      }
    }
    return jobs.length
  }

  /**
   * @param {object} ctx
   * @param {{ id: string, kind: string, ref_id: string|null }} job
   */
  async function execute(ctx, job) {
    if (job.ref_id === null) throw new Error(`vk_jobs.ref_id missing for ${job.kind}, id: ${job.id}`)
    const task_id = job.ref_id
    if (TASK_ACTION_KINDS.includes(job.kind)) return void await tasks.actAsWorker(ctx, { task_id, action: { type: job.kind } })
    if (WARN_KINDS.includes(job.kind)) return tasks.warn(ctx, { task_id, kind: job.kind })
    if (job.kind === 'transfer_expire') return tasks.expireTransfers(ctx, { task_id })
    if (job.kind === 'review_auto_accept') {
      const row = await db('vk_tasks').where({ id: task_id }).first('template', 'status')
      if (row === undefined) throw new Error(`vk_tasks row missing for review_auto_accept, id: ${task_id}`)
      // 擂台赛到期要由大模型按清单选赢家；大模型尚未装配，这里明确失败，不替它做决定（发布方仍可手动选赢家）。
      if (row.template === 'contest' && row.status === 'reviewing') throw new Error(`review_auto_accept for a contest needs the LLM judge, which is not assembled yet, id: ${task_id}`)
      return void await tasks.actAsWorker(ctx, { task_id, action: { type: 'review_auto_accept', winner_agent_id: null } })
    }
    // 其余类型（仲裁意见、Webhook、日报、对账）在大模型装配后与 W4 接入。
    throw new Error(`Unhandled job kind: ${job.kind}, id: ${job.id}`)
  }

  return { runDue }
}