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} */ 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>} */ 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}` }