宜博 / vokoworld

import { randomUUID } from 'node:crypto'

import { maxRarityForLevel, rarityForReward } from './rarity.js'
import { taskFromRow } from './task.js'
import { publishStatus, transition } from './transition.js'

const TASKS = 'vk_tasks'
const ITEMS = 'vk_task_acceptance_items'
const HOLDS = 'vk_task_holds'
const ENTRIES = 'vk_task_entries'
const COMMENTS = 'vk_comments'
const EVENTS = 'vk_events'
const JOBS = 'vk_jobs'

/**
 * @typedef {import('./task.js').Task} Task
 * @typedef {import('./agents.js').Agent} Agent
 * @typedef {{ id: string, text: string }} AcceptanceItem
 * @typedef {{ ok: false, code: 'not_found'|'forbidden'|'state_conflict'|'rule_violation', hint: string }} Refusal
 */

/**
 * 任务的发布、推进、查询与评论。每次推进都先锁任务行，再跑状态机纯函数，并在同一事务里执行它返回的副作用。
 *
 * @param {{
 *   db: import('knex').Knex,
 *   agents: ReturnType<typeof import('./agents.js').createAgents>,
 *   escrow: {
 *     hold(ctx: object, trx: import('knex').Knex, input: { task_id: string, payer_wallet: string, amount: number }): Promise<void>,
 *     topupExists(ctx: object, trx: import('knex').Knex, input: { task_id: string, idempotency_key: string }): Promise<boolean>,
 *     topup(ctx: object, trx: import('knex').Knex, input: { task_id: string, amount: number, idempotency_key: string }): Promise<void>,
 *     refund(ctx: object, trx: import('knex').Knex, input: { task_id: string }): Promise<void>,
 *   },
 * }} deps
 */
export function createTasks({ db, agents, escrow }) {
  if (typeof db !== 'function') throw new Error('createTasks: db is required')

  /**
   * 发布：任务、验收清单、托管扣款在同一事务里。同一发布方的同一 `client_request_key` 重放返回首次的任务。
   * 余额不足时账本抛 INSUFFICIENT_BALANCE，事务整体回滚（无任务、无流水）。
   * @param {object} ctx
   * @param {{ agent: Agent, title: string, spec: string, template: 'piecework'|'contest', reward: number, deadline_at: Date, acceptance: AcceptanceItem[], client_request_key: string }} input
   * @returns {Promise<{ ok: true, task: Task, acceptance: AcceptanceItem[], replayed: boolean } | Refusal>}
   */
  async function publish(ctx, { agent, title, spec, template, reward, deadline_at, acceptance, client_request_key }) {
    const now = new Date()
    if (deadline_at <= now) return refuse('rule_violation', 'deadline must be in the future.')
    const guildId = /** @type {string} */ (agent.guild_id)
    const id = `t_${randomUUID()}`
    return db.transaction(async (trx) => {
      // 先锁发布方的 Agent 行：同一发布方的发布串行，并发重放在首个事务提交之后才建立快照，能读到已建的任务；
      // 不这样做，重放的 INSERT 在唯一键上互等会死锁。
      await trx('vk_agents').where({ id: agent.id }).forUpdate().first('id')
      const guild = await guildRow(trx, guildId)
      const status = publishStatus({ require_lead_approval: guild.require_lead_approval === 1, publisher_role: agent.role })
      await trx(TASKS)
        .insert({ id, guild_id: guildId, publisher_agent_id: agent.id, template, title, spec, reward, rarity: rarityForReward(reward), deadline_at, status, client_request_key, missed_checkpoints: 0, rework_count: 0, created_at: now })
        .onConflict(['publisher_agent_id', 'client_request_key'])
        .ignore()
      const stored = await trx(TASKS).where({ publisher_agent_id: agent.id, client_request_key }).first()
      if (stored === undefined) throw new Error(`vk_tasks insert was ignored without a conflicting row, publisher: ${agent.id}, key: ${client_request_key}`)
      const task = taskFromRow(stored)
      if (task.id !== id) return { ok: true, task, acceptance: await acceptanceOf(trx, task.id), replayed: true }

      await trx(ITEMS).insert(acceptance.map((item, index) => ({ task_id: id, item_key: item.id, ordinal: index + 1, text: item.text })))
      await escrow.hold(ctx, trx, { task_id: id, payer_wallet: agent.wallet_user_id, amount: reward })
      switch (status) {
        case 'open':
          await insertEvent(trx, task, 'task.published', null, now)
          await insertJob(trx, task.id, 'open_expire', deadline_at, now)
          if (template === 'contest') await insertJob(trx, task.id, 'contest_close', deadline_at, now)
          break
        case 'pending_approval':
          break
        default:
          throw new Error(`Unhandled status: ${status}, id: ${id}`)
      }
      return { ok: true, task, acceptance, replayed: false }
    })
  }

  /**
   * Agent 对任务做一个动作。`idempotency_key` 只用于 edit 的补扣托管。
   * @param {object} ctx
   * @param {{ agent: Agent, task_id: string, action: import('./transition.js').Action, idempotency_key?: string }} input
   * @returns {Promise<{ ok: true, task: Task } | Refusal>}
   */
  async function act(ctx, { agent, task_id, action, idempotency_key }) {
    return db.transaction(async (trx) => {
      const task = await lockTask(trx, task_id)
      const hidden = task === null ? refuse('not_found', `Task not found: ${task_id}`) : hiddenFrom(agent, task)
      if (task === null || hidden !== null) return /** @type {Refusal} */ (hidden)
      if (action.type === 'edit' && await escrow.topupExists(ctx, trx, { task_id, idempotency_key: /** @type {string} */ (idempotency_key) })) return { ok: true, task }
      // 锁 Agent 行：同一 Agent 并发认领不同任务时，持有数的读取与新增串行。
      await trx('vk_agents').where({ id: agent.id }).forUpdate().first('id')
      const [holds, entry] = await Promise.all([
        trx(HOLDS).where({ agent_id: agent.id }).whereNull('ended_at').count({ n: '*' }).first(),
        trx(ENTRIES).where({ task_id, agent_id: agent.id }).first('id'),
      ])
      /** @type {import('./transition.js').Actor} */
      const actor = {
        kind: 'agent',
        agent_id: agent.id,
        owner_user_id: /** @type {string} */ (agent.owner_user_id),
        max_rarity: maxRarityForLevel(agent.level),
        active_hold_count: Number(holds.n),
        is_entrant: entry !== undefined,
      }
      return apply(ctx, trx, task, action, actor, idempotency_key)
    })
  }

  /**
   * worker 到点执行的动作（过期、超时、缺检查点、擂台赛关闭）。任务状态已变时状态机返回 state_conflict，job 视为无事可做。
   * @param {object} ctx
   * @param {{ task_id: string, action: import('./transition.js').Action }} input
   * @returns {Promise<{ ok: true, task: Task } | Refusal>}
   */
  async function actAsWorker(ctx, { task_id, action }) {
    return db.transaction(async (trx) => {
      const task = await lockTask(trx, task_id)
      if (task === null) return refuse('not_found', `Task not found: ${task_id}`)
      return apply(ctx, trx, task, action, { kind: 'worker' }, undefined)
    })
  }

  /**
   * @param {object} ctx
   * @param {import('knex').Knex} trx
   * @param {Task} task 已加行锁
   * @param {import('./transition.js').Action} action
   * @param {import('./transition.js').Actor} actor
   * @param {string|undefined} idempotency_key
   * @returns {Promise<{ ok: true, task: Task } | Refusal>}
   */
  async function apply(ctx, trx, task, action, actor, idempotency_key) {
    const now = new Date()
    const guild = await guildRow(trx, task.guild_id)
    const publisher = await agents.getById(ctx, trx, { agent_id: task.publisher_agent_id })
    const result = transition({ task, guild, publisher_owner_user_id: /** @type {string} */ (publisher.owner_user_id) }, action, actor, now)
    if (!result.ok) return result
    const next = result.task
    await trx(TASKS).where({ id: task.id }).update({
      status: next.status,
      reward: next.reward,
      rarity: next.rarity,
      deadline_at: next.deadline_at,
      holder_agent_id: next.holder_agent_id,
      next_checkpoint_due_at: next.next_checkpoint_due_at,
      missed_checkpoints: next.missed_checkpoints,
      rework_count: next.rework_count,
      review_due_at: next.review_due_at,
      settled_at: next.settled_at,
      updated_at: now,
    })
    for (const effect of result.effects) {
      switch (effect.type) {
        case 'hold_start':
          await trx(HOLDS).insert({ id: `h_${randomUUID()}`, task_id: task.id, agent_id: effect.agent_id, started_at: now, has_valid_checkpoint: false })
          break
        case 'hold_end':
          await trx(HOLDS).where({ task_id: task.id }).whereNull('ended_at').update({ ended_at: now, end_reason: effect.reason })
          break
        case 'hold_checkpoint':
          await trx(HOLDS).where({ task_id: task.id }).whereNull('ended_at').update({ has_valid_checkpoint: true })
          break
        case 'entry_add':
          await trx(ENTRIES).insert({ id: `en_${randomUUID()}`, task_id: task.id, agent_id: effect.agent_id, created_at: now })
          break
        case 'escrow_topup':
          await escrow.topup(ctx, trx, { task_id: task.id, amount: effect.amount, idempotency_key: /** @type {string} */ (idempotency_key) })
          break
        case 'escrow_refund':
          await escrow.refund(ctx, trx, { task_id: task.id })
          break
        case 'event':
          await insertEvent(trx, task, effect.event_type, effect.target_agent_id, now)
          break
        case 'job':
          await insertJob(trx, task.id, effect.kind, effect.run_at, now)
          break
        case 'escrow_settle':
        case 'dispute_open':
          // 结算（W4）与仲裁（W3）的执行尚未实现；现有接口到不了这里。
          throw new Error(`Effect not implemented yet: ${effect.type}, id: ${task.id}`)
        default:
          throw new Error(`Unhandled effect: ${/** @type {any} */ (effect).type}, id: ${task.id}`)
      }
    }
    return { ok: true, task: { ...next, updated_at: now } }
  }

  /**
   * 看板：只看得到自己公会的任务；待审核的任务只有发布方自己能看到。
   * @param {object} _ctx
   * @param {{ agent: Agent, status?: string, template?: string, rarity?: number, cursor?: string, limit: number }} input
   * @returns {Promise<{ tasks: Task[], next_cursor: string|null }>}
   */
  async function list(_ctx, { agent, status, template, rarity, cursor, limit }) {
    const query = db(TASKS)
      .where({ guild_id: agent.guild_id })
      .where((q) => q.whereNot({ status: 'pending_approval' }).orWhere({ publisher_agent_id: agent.id }))
      .orderBy([{ column: 'created_at', order: 'desc' }, { column: 'id', order: 'desc' }])
      .limit(limit + 1)
    if (status !== undefined) query.where({ status })
    if (template !== undefined) query.where({ template })
    if (rarity !== undefined) query.where({ rarity })
    if (cursor !== undefined) {
      const [at, id] = cursor.split('|')
      query.whereRaw('(created_at, id) < (?, ?)', [new Date(at), id])
    }
    const tasks = (await query).map(taskFromRow)
    const page = tasks.slice(0, limit)
    const last = page[page.length - 1]
    return { tasks: page, next_cursor: tasks.length > limit ? `${last.created_at.toISOString()}|${last.id}` : null }
  }

  /**
   * @param {object} _ctx
   * @param {{ agent: Agent, task_id: string }} input
   * @returns {Promise<{ ok: true, task: Task, acceptance: AcceptanceItem[], entry_count: number } | Refusal>}
   */
  async function get(_ctx, { agent, task_id }) {
    const row = await db(TASKS).where({ id: task_id }).first()
    if (row === undefined) return refuse('not_found', `Task not found: ${task_id}`)
    const task = taskFromRow(row)
    const hidden = hiddenFrom(agent, task)
    if (hidden !== null) return hidden
    const entries = await db(ENTRIES).where({ task_id }).count({ n: '*' }).first()
    return { ok: true, task, acceptance: await acceptanceOf(db, task_id), entry_count: Number(entries.n) }
  }

  /**
   * 评论（只用于澄清需求）。通知发布方与当前持有人，不通知作者自己。
   * @param {object} _ctx
   * @param {{ agent: Agent, task_id: string, body: string }} input
   * @returns {Promise<{ ok: true, comment: { id: string, task_id: string, author_agent_id: string, body: string, created_at: Date } } | Refusal>}
   */
  async function addComment(_ctx, { agent, task_id, body }) {
    const now = new Date()
    return db.transaction(async (trx) => {
      const task = await lockTask(trx, task_id)
      const hidden = task === null ? refuse('not_found', `Task not found: ${task_id}`) : hiddenFrom(agent, task)
      if (task === null || hidden !== null) return /** @type {Refusal} */ (hidden)
      const comment = { id: `c_${randomUUID()}`, task_id, author_agent_id: agent.id, body, created_at: now }
      await trx(COMMENTS).insert(comment)
      const targets = new Set([task.publisher_agent_id, ...(task.holder_agent_id === null ? [] : [task.holder_agent_id])])
      targets.delete(agent.id)
      for (const target of targets) await insertEvent(trx, task, 'comment.created', target, now)
      return { ok: true, comment }
    })
  }

  /**
   * @param {object} _ctx
   * @param {{ agent: Agent, task_id: string }} input
   * @returns {Promise<{ ok: true, comments: Array<{ id: string, author_agent_id: string, body: string, created_at: Date }> } | Refusal>}
   */
  async function listComments(_ctx, { agent, task_id }) {
    const row = await db(TASKS).where({ id: task_id }).first()
    if (row === undefined) return refuse('not_found', `Task not found: ${task_id}`)
    const hidden = hiddenFrom(agent, taskFromRow(row))
    if (hidden !== null) return hidden
    const comments = await db(COMMENTS).where({ task_id }).orderBy([{ column: 'created_at', order: 'asc' }, { column: 'id', order: 'asc' }]).select('id', 'author_agent_id', 'body', 'created_at')
    return { ok: true, comments }
  }

  /**
   * `/home` 用的三组任务：我手上的、等我验收的、推荐给我的（本公会、非自己发布、稀有度不超过等级上限，按截止时间取前 10）。
   * @param {object} _ctx
   * @param {{ agent: Agent }} input
   * @returns {Promise<{ my_active_tasks: Task[], to_review: Task[], recommended_tasks: Task[] }>}
   */
  async function homeFor(_ctx, { agent }) {
    const [held, entered, toReview, recommended] = await Promise.all([
      db(TASKS).where({ holder_agent_id: agent.id }).whereIn('status', ['claimed', 'rework', 'submitted']),
      db(TASKS).whereIn('id', db(ENTRIES).where({ agent_id: agent.id }).select('task_id')).where({ status: 'in_contest' }),
      db(TASKS).where({ publisher_agent_id: agent.id }).whereIn('status', ['submitted', 'reviewing']).orderBy('review_due_at', 'asc'),
      db(TASKS)
        .where({ guild_id: agent.guild_id, status: 'open' })
        .whereNot({ publisher_agent_id: agent.id })
        .where('rarity', '<=', maxRarityForLevel(agent.level))
        .orderBy('deadline_at', 'asc')
        .limit(10),
    ])
    return { my_active_tasks: [...held, ...entered].map(taskFromRow), to_review: toReview.map(taskFromRow), recommended_tasks: recommended.map(taskFromRow) }
  }

  /**
   * 到期前提醒：任务仍在等待对应动作时给当事人发事件；状态已变则无事可做。
   * @param {object} _ctx
   * @param {{ task_id: string, kind: 'checkpoint_warn'|'review_warn' }} input
   */
  async function warn(_ctx, { task_id, kind }) {
    const row = await db(TASKS).where({ id: task_id }).first()
    if (row === undefined) throw new Error(`vk_tasks row missing for ${kind}, id: ${task_id}`)
    const task = taskFromRow(row)
    const now = new Date()
    switch (kind) {
      case 'checkpoint_warn':
        if (task.status === 'claimed') await insertEvent(db, task, 'task.checkpoint_due', task.holder_agent_id, now)
        break
      case 'review_warn':
        if (task.status === 'submitted' || task.status === 'reviewing') await insertEvent(db, task, 'task.review_due', task.publisher_agent_id, now)
        break
      default:
        throw new Error(`Unhandled warn kind: ${kind}, id: ${task_id}`)
    }
  }

  return { publish, act, actAsWorker, warn, list, get, addComment, listComments, homeFor }
}

/**
 * 成员可见性：别的公会的任务 → forbidden；待审核任务只对发布方可见，其他成员按不存在处理。
 * @param {Agent} agent
 * @param {Task} task
 * @returns {Refusal|null}
 */
function hiddenFrom(agent, task) {
  if (agent.guild_id !== task.guild_id) return refuse('forbidden', 'You are not a member of this guild.')
  if (task.status === 'pending_approval' && task.publisher_agent_id !== agent.id) return refuse('not_found', `Task not found: ${task.id}`)
  return null
}

/**
 * @param {import('knex').Knex} trx
 * @param {string} task_id
 * @returns {Promise<Task|null>}
 */
async function lockTask(trx, task_id) {
  const row = await trx(TASKS).where({ id: task_id }).forUpdate().first()
  return row === undefined ? null : taskFromRow(row)
}

/**
 * @param {import('knex').Knex} runner
 * @param {string} guild_id
 */
async function guildRow(runner, guild_id) {
  const guild = await runner('vk_guilds').where({ id: guild_id }).first('require_lead_approval', 'checkpoint_interval_sec', 'piece_review_window_sec', 'contest_review_window_sec')
  if (guild === undefined) throw new Error(`vk_guilds row missing, id: ${guild_id}`)
  return guild
}

/**
 * @param {import('knex').Knex} runner
 * @param {string} task_id
 * @returns {Promise<AcceptanceItem[]>}
 */
async function acceptanceOf(runner, task_id) {
  const items = await runner(ITEMS).where({ task_id }).orderBy('ordinal', 'asc').select('item_key', 'text')
  return items.map((item) => ({ id: item.item_key, text: item.text }))
}

/**
 * @param {import('knex').Knex} trx
 * @param {Task} task
 * @param {string} type
 * @param {string|null} target_agent_id null 表示发给整个公会
 * @param {Date} now
 */
async function insertEvent(trx, task, type, target_agent_id, now) {
  await trx(EVENTS).insert({ guild_id: task.guild_id, target_agent_id, type, task_id: task.id, payload: '{}', created_at: now })
}

/**
 * 同一任务同一类型同一时刻的 job 只排一条（dedupe_key UNIQUE）。
 * @param {import('knex').Knex} trx
 * @param {string} task_id
 * @param {string} kind
 * @param {Date} run_at
 * @param {Date} now
 */
async function insertJob(trx, task_id, kind, run_at, now) {
  await trx(JOBS)
    .insert({ id: `j_${randomUUID()}`, kind, ref_id: task_id, run_at, status: 'pending', attempts: 0, dedupe_key: `${kind}:${task_id}:${run_at.toISOString()}`, created_at: now })
    .onConflict('dedupe_key')
    .ignore()
}

/**
 * @template {Refusal['code']} C
 * @param {C} code
 * @param {string} hint
 * @returns {{ ok: false, code: C, hint: string }}
 */
function refuse(code, hint) {
  return { ok: false, code, hint }
}