import { randomUUID } from 'node:crypto'

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>,
 *     autoSelectWinner(ctx: object, input: { task_id: string }): Promise<void>,
 *     arbitrate(ctx: object, input: { task_id: string }): Promise<void>,
 *   },
 *   help: { notify(ctx: object, input: { help_id: string }): Promise<void> },
 *   sendDigest(ctx: object, input: { user_id: string }): Promise<void>,
 *   runDailyGame(ctx: object): Promise<unknown>,
 *   conservationProblems(): Promise<Array<{ rule: string, id: string, detail: string }>>,
 *   logger: { error(message: string, fields: object): void },
 * }} deps
 */
export function createJobs({ db, tasks, help, sendDigest, runDailyGame, conservationProblems, 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.kind === 'verify_conservation') {
      // 每日对账：有任何不守恒就让 job 失败（记错误日志即告警），并附上出问题的数据 id。
      const problems = await conservationProblems()
      if (problems.length > 0) throw new Error(`conservation check failed: ${JSON.stringify(problems.slice(0, 20))}`)
      return
    }
    if (job.kind === 'daily_game') return void await runDailyGame(ctx)
    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 (job.kind === 'daily_digest') return sendDigest(ctx, { user_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')
      if (row === undefined) throw new Error(`vk_tasks row missing for review_auto_accept, id: ${task_id}`)
      // 计件活到期直接结算；擂台赛由大模型按清单选赢家。
      if (row.template === 'contest') return tasks.autoSelectWinner(ctx, { task_id })
      return void await tasks.actAsWorker(ctx, { task_id, action: { type: 'review_auto_accept', winner_agent_id: null } })
    }
    if (job.kind === 'dispute_arbitrate') return tasks.arbitrate(ctx, { task_id })
    if (job.kind === 'help_notify') return help.notify(ctx, { help_id: job.ref_id })
    // 其余类型（Webhook、日报、对账）在 W4 接入。
    throw new Error(`Unhandled job kind: ${job.kind}, id: ${job.id}`)
  }

  /**
   * 排一条 job；同一类型、同一对象、同一时刻只排一次（dedupe_key UNIQUE），重复调用无害。
   * @param {object} _ctx
   * @param {{ kind: string, ref_id: string|null, run_at: Date }} input
   */
  async function enqueue(_ctx, { kind, ref_id, run_at }) {
    await db(JOBS)
      .insert({ id: `j_${randomUUID()}`, kind, ref_id, run_at, status: 'pending', attempts: 0, dedupe_key: `${kind}:${ref_id === null ? '' : ref_id}:${run_at.toISOString()}`, created_at: new Date() })
      .onConflict('dedupe_key')
      .ignore()
  }

  /**
   * 确保下一次每日对账（UTC 01:00）与每日称号结算（UTC 00:10）已经排上。
   * @param {object} ctx
   */
  async function ensureDaily(ctx) {
    const now = new Date()
    for (const [kind, hour, minute] of /** @type {const} */ ([['verify_conservation', 1, 0], ['daily_game', 0, 10]])) {
      const next = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate(), hour, minute))
      if (next <= now) next.setUTCDate(next.getUTCDate() + 1)
      await enqueue(ctx, { kind, ref_id: null, run_at: next })
    }
  }

  return { runDue, enqueue, ensureDaily }
}