宜博 / vokoworld

import { randomUUID } from 'node:crypto'
import { createBillingRepo, createLedger } from '@drama/billing'

const ESCROWS = 'vk_escrows'
const PAYOUTS = 'vk_escrow_payouts'
const SOURCE = 'voko'
const REF_TYPE = 'vk_task'
const PAYOUT_REASON = Object.freeze({ winner: 'vk_task_payout', relay_share: 'vk_relay_share' })

/**
 * 托管：发布时从发布方钱包扣款并记 held，结算时发给接单方，退回时原路发还。
 * 所有方法都在调用方的事务（`trx`）里动账，业务表与账本一起提交或回滚；幂等键由任务唯一决定。
 * 余额不足时账本抛 `BillingError(INSUFFICIENT_BALANCE)`，整个事务回滚。
 *
 * @param {{ db: import('knex').Knex }} deps
 */
export function createEscrow({ db }) {
  if (typeof db !== 'function') throw new Error('createEscrow: db is required')

  /** @param {import('knex').Knex} trx */
  const ledgerIn = (trx) => createLedger(createBillingRepo(db, trx))

  /**
   * @param {object} _ctx
   * @param {import('knex').Knex} trx
   * @param {{ task_id: string, payer_wallet: string, amount: number }} input
   */
  async function hold(_ctx, trx, { task_id, payer_wallet, amount }) {
    const deducted = await ledgerIn(trx).deduct({ user_id: payer_wallet, amount, source: SOURCE, reason: 'vk_escrow_hold', ref_type: REF_TYPE, ref_id: task_id, idempotency_key: `vk:escrow:${task_id}` })
    await trx(ESCROWS).insert({ id: `es_${randomUUID()}`, task_id, payer_wallet, amount, status: 'held', deduct_ledger_id: deducted.ledger_id, created_at: new Date() })
  }

  /**
   * 加赏金的补扣是否已经做过（同一 Idempotency-Key 的重放）。
   * @param {object} _ctx
   * @param {import('knex').Knex} trx
   * @param {{ task_id: string, idempotency_key: string }} input
   */
  async function topupExists(_ctx, trx, { task_id, idempotency_key }) {
    return (await createBillingRepo(db, trx).findLedgerByIdempotencyKey(topupKey(task_id, idempotency_key))) !== null
  }

  /**
   * @param {object} _ctx
   * @param {import('knex').Knex} trx
   * @param {{ task_id: string, amount: number, idempotency_key: string }} input
   */
  async function topup(_ctx, trx, { task_id, amount, idempotency_key }) {
    const escrow = await heldEscrow(trx, task_id)
    await ledgerIn(trx).deduct({ user_id: escrow.payer_wallet, amount, source: SOURCE, reason: 'vk_escrow_hold', ref_type: REF_TYPE, ref_id: task_id, idempotency_key: topupKey(task_id, idempotency_key) })
    await trx(ESCROWS).where({ id: escrow.id }).increment('amount', amount)
  }

  /**
   * 全额退回发布方：escrow → refunded，并记一条 refund 发放明细。
   * @param {object} _ctx
   * @param {import('knex').Knex} trx
   * @param {{ task_id: string }} input
   */
  async function refund(_ctx, trx, { task_id }) {
    const escrow = await heldEscrow(trx, task_id)
    const now = new Date()
    const granted = await ledgerIn(trx).grant({
      user_id: escrow.payer_wallet,
      amount: escrow.amount,
      source: SOURCE,
      credit_kind: 'earned',
      reason: 'vk_escrow_refund',
      ref_type: REF_TYPE,
      ref_id: task_id,
      idempotency_key: `vk:payout:${task_id}:${escrow.payer_wallet}:refund`,
    })
    await trx(PAYOUTS).insert({ id: `ep_${randomUUID()}`, escrow_id: escrow.id, task_id, recipient_wallet: escrow.payer_wallet, amount: escrow.amount, kind: 'refund', ledger_id: granted.ledger_id, created_at: now })
    await trx(ESCROWS).where({ id: escrow.id }).update({ status: 'refunded', closed_at: now })
  }

  /**
   * 结算：按给定的发放明细把托管发给接单方（含接力分成），escrow → released。
   * 各笔之和必须等于托管金额，否则 throw（宁可 500，不写脏账）。
   * @param {object} _ctx
   * @param {import('knex').Knex} trx
   * @param {{ task_id: string, payouts: Array<{ wallet_user_id: string, amount: number, kind: 'winner'|'relay_share' }> }} input
   */
  async function settle(_ctx, trx, { task_id, payouts }) {
    const escrow = await heldEscrow(trx, task_id)
    const total = payouts.reduce((sum, payout) => sum + payout.amount, 0)
    if (total !== escrow.amount) throw new Error(`vk_escrows settle payouts ${total} do not equal escrow amount ${escrow.amount}, id: ${escrow.id}`)
    const now = new Date()
    // 固定按钱包 id 排序动账，避免两笔结算互相等待对方的钱包行。
    for (const payout of [...payouts].sort((a, b) => (a.wallet_user_id < b.wallet_user_id ? -1 : 1))) {
      const granted = await ledgerIn(trx).grant({
        user_id: payout.wallet_user_id,
        amount: payout.amount,
        source: SOURCE,
        credit_kind: 'earned',
        reason: PAYOUT_REASON[payout.kind],
        ref_type: REF_TYPE,
        ref_id: task_id,
        idempotency_key: `vk:payout:${task_id}:${payout.wallet_user_id}:${payout.kind}`,
      })
      await trx(PAYOUTS).insert({ id: `ep_${randomUUID()}`, escrow_id: escrow.id, task_id, recipient_wallet: payout.wallet_user_id, amount: payout.amount, kind: payout.kind, ledger_id: granted.ledger_id, created_at: now })
    }
    await trx(ESCROWS).where({ id: escrow.id }).update({ status: 'released', closed_at: now })
  }

  /**
   * 钱包视图：可用 = 账本余额；冻结 = 该钱包名下 held 托管之和。
   * @param {object} _ctx
   * @param {{ wallet_user_id: string }} input
   * @returns {Promise<{ available: number, held: number }>}
   */
  async function walletOf(_ctx, { wallet_user_id }) {
    const [available, heldRow] = await Promise.all([
      createBillingRepo(db).getBalance(wallet_user_id),
      db(ESCROWS).where({ payer_wallet: wallet_user_id, status: 'held' }).sum({ held: 'amount' }).first(),
    ])
    return { available, held: Number(heldRow.held) }
  }

  /**
   * 该钱包今天（UTC 日）已冻结进托管的积分，用于单日支出上限。
   * @param {object} _ctx
   * @param {import('knex').Knex} trx
   * @param {{ wallet_user_id: string, now: Date }} input
   * @returns {Promise<number>}
   */
  async function spentToday(_ctx, trx, { wallet_user_id, now }) {
    const dayStart = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()))
    const row = await trx('billing_ledger').where({ user_id: wallet_user_id, source: SOURCE, reason: 'vk_escrow_hold' }).where('created_at', '>=', dayStart).sum({ spent: 'amount' }).first()
    return -Number(row.spent)
  }

  /**
   * 钱包流水，最近的在前。
   * @param {object} _ctx
   * @param {{ wallet_user_id: string, limit: number }} input
   * @returns {Promise<Array<{ id: string, type: string, amount: number, reason: string, ref_type: string|null, ref_id: string|null, created_at: Date }>>}
   */
  async function ledgerOf(_ctx, { wallet_user_id, limit }) {
    return db('billing_ledger').where({ user_id: wallet_user_id }).orderBy([{ column: 'created_at', order: 'desc' }, { column: 'id', order: 'desc' }]).limit(limit).select('id', 'type', 'amount', 'reason', 'ref_type', 'ref_id', 'created_at')
  }

  return { hold, topupExists, topup, refund, settle, walletOf, ledgerOf, spentToday }
}

/**
 * 取任务的 held 托管并加行锁；不存在或已关闭都是不变量被破坏。
 * @param {import('knex').Knex} trx
 * @param {string} task_id
 * @returns {Promise<{ id: string, payer_wallet: string, amount: number }>}
 */
async function heldEscrow(trx, task_id) {
  const row = await trx(ESCROWS).where({ task_id }).forUpdate().first()
  if (row === undefined || row.status !== 'held') throw new Error(`vk_escrows expected held escrow, task_id: ${task_id}, status: ${row === undefined ? 'missing' : row.status}`)
  if (!Number.isSafeInteger(row.amount) || row.amount <= 0) throw new Error(`vk_escrows.amount invalid, id: ${row.id}`)
  return { id: row.id, payer_wallet: row.payer_wallet, amount: row.amount }
}

/**
 * @param {string} task_id
 * @param {string} idempotency_key
 */
function topupKey(task_id, idempotency_key) {
  return `vk:escrow_topup:${task_id}:${idempotency_key}`
}