diff --git a/apps/voko-server/package.json b/apps/voko-server/package.json index 3c1f852..e7402bc 100644 --- a/apps/voko-server/package.json +++ b/apps/voko-server/package.json @@ -13,7 +13,12 @@ "db:status": "node src/db/migration-status.js", "admin:grant": "node tools/grant-admin.mjs", "verify:skeleton": "node tools/scenarios/voko-alpha/skeleton.mjs", - "verify:f0": "node tools/scenarios/voko-alpha/f0-onboarding.mjs" + "verify:f0": "node tools/scenarios/voko-alpha/f0-onboarding.mjs", + "verify:f1": "node tools/scenarios/voko-alpha/f1-publish.mjs", + "verify:f2": "node tools/scenarios/voko-alpha/f2-execute.mjs", + "verify:timers": "node tools/scenarios/voko-alpha/timers.mjs", + "verify:concurrency": "node tools/scenarios/voko-alpha/concurrency.mjs", + "verify:conservation": "node tools/scenarios/voko-alpha/verify-conservation.mjs" }, "dependencies": { "@drama/auth": "workspace:*", @@ -22,6 +27,7 @@ "@drama/db": "workspace:*", "@drama/email": "workspace:*", "@drama/observability": "workspace:*", + "@drama/ratelimit": "workspace:*", "@drama/server-kit": "workspace:*", "@voko/access": "workspace:*", "@voko/clearing": "workspace:*", diff --git a/apps/voko-server/src/assembly/runtime.js b/apps/voko-server/src/assembly/runtime.js index 7640b37..4b2a0bf 100644 --- a/apps/voko-server/src/assembly/runtime.js +++ b/apps/voko-server/src/assembly/runtime.js @@ -9,7 +9,9 @@ import { createBillingRepo, createLedger } from '@drama/billing' import { createDb } from '@drama/db' import { createLogger } from '@drama/observability' import { createAccessRequests, createAgentAccess } from '@voko/access' -import { createAgents, createGuilds } from '@voko/exchange' +import { createMemoryStore, createRateLimiter } from '@drama/ratelimit' +import { createEscrow } from '@voko/clearing' +import { createAgents, createEvents, createGuilds, createJobs, createTasks } from '@voko/exchange' import { readEnv } from '../env.js' import { createEmailPorts } from './email.js' @@ -44,6 +46,19 @@ export async function createRuntime({ service }) { const access = createAccessRequests({ db, notifier: accessNotifier }) const agents = createAgents() const guilds = createGuilds({ db }) + const escrow = createEscrow({ db }) + const tasks = createTasks({ db, agents, escrow }) + const events = createEvents({ db }) + const jobs = createJobs({ db, tasks, logger }) + // 公开 API 按 Agent 限流(接入说明:读 120/分、写 40/分、发任务 30/小时)。单进程部署用内存 store,多实例时换 MySQL 或 Redis store。 + const rateLimiter = createRateLimiter({ + store: createMemoryStore(), + rules: [ + { endpoint: 'read', by: 'anon', capacity: 120, refillPerSec: 120 / 60 }, + { endpoint: 'write', by: 'anon', capacity: 40, refillPerSec: 40 / 60 }, + { endpoint: 'publish', by: 'anon', capacity: 30, refillPerSec: 30 / 3600 }, + ], + }) const agentAccess = createAgentAccess({ db, agents, @@ -78,7 +93,7 @@ export async function createRuntime({ service }) { return { env, logger, - services: { db, auth, userRepo, access, agents, guilds, agentAccess, register, oauthEnabled: env.oauth.googleClientId !== null }, + services: { db, auth, userRepo, access, agents, guilds, agentAccess, escrow, tasks, events, jobs, rateLimiter, register, oauthEnabled: env.oauth.googleClientId !== null }, async close() { await db.destroy() }, diff --git a/apps/voko-server/src/routes/v1.js b/apps/voko-server/src/routes/v1.js index 620f07d..37f510e 100644 --- a/apps/voko-server/src/routes/v1.js +++ b/apps/voko-server/src/routes/v1.js @@ -1,4 +1,5 @@ import * as v from 'valibot' +import { TASK_STATUSES } from '@voko/exchange' const Skills = v.pipe(v.array(v.pipe(v.string(), v.minLength(1), v.maxLength(40))), v.maxLength(20)) const Description = v.pipe(v.string(), v.minLength(1), v.maxLength(1000)) @@ -8,6 +9,38 @@ const RegisterBody = v.object({ skills: Skills, owner_email: v.optional(v.pipe(v.string(), v.email(), v.maxLength(320))), }) +const TaskParams = v.object({ id: v.pipe(v.string(), v.minLength(1), v.maxLength(64)) }) +const IsoDate = v.pipe(v.string(), v.isoTimestamp(), v.transform((value) => new Date(value))) +const Reward = v.pipe(v.number(), v.safeInteger(), v.minValue(1), v.maxValue(1_000_000_000)) +const PublishBody = v.object({ + title: v.pipe(v.string(), v.minLength(1), v.maxLength(200)), + spec: v.pipe(v.string(), v.minLength(1), v.maxLength(20000)), + template: v.picklist(['piecework', 'contest']), + reward: Reward, + deadline: IsoDate, + acceptance: v.pipe( + v.array(v.object({ id: v.pipe(v.string(), v.minLength(1), v.maxLength(32)), text: v.pipe(v.string(), v.minLength(1), v.maxLength(1000)) })), + v.minLength(1), + v.maxLength(5), + v.check((items) => new Set(items.map((item) => item.id)).size === items.length, 'acceptance ids must be unique'), + ), +}) +const EditBody = v.object({ add_reward: v.optional(Reward), deadline: v.optional(IsoDate) }) +const ListQuery = v.object({ + status: v.optional(v.picklist(TASK_STATUSES)), + template: v.optional(v.picklist(['piecework', 'contest'])), + rarity: v.optional(v.pipe(v.string(), v.transform(Number), v.integer(), v.minValue(1), v.maxValue(10))), + cursor: v.optional(v.pipe(v.string(), v.regex(/^[^|]+\|[^|]+$/))), + limit: v.optional(v.pipe(v.string(), v.transform(Number), v.integer(), v.minValue(1), v.maxValue(100)), '50'), +}) +const CommentBody = v.object({ body: v.pipe(v.string(), v.minLength(1), v.maxLength(4000)) }) +const EventsQuery = v.object({ since: v.optional(v.pipe(v.string(), v.transform(Number), v.integer(), v.minValue(0)), '0') }) +const IdempotencyKey = v.pipe(v.string('Idempotency-Key header is required'), v.minLength(1), v.maxLength(100)) + +/** 业务拒绝 → HTTP 状态与公开错误码(接入说明「错误码」)。 */ +const REFUSAL_STATUS = Object.freeze({ not_found: 404, forbidden: 403, state_conflict: 409, rule_violation: 422 }) +const SKILL_VERSION = '0.1.0' + const UpdateMeBody = v.object({ description: v.optional(Description), skills: v.optional(Skills) }) /** @@ -16,7 +49,7 @@ const UpdateMeBody = v.object({ description: v.optional(Description), skills: v. * @param {Record} services */ export function registerV1Routes(app, services) { - const { agentAccess, agents, db } = services + const { agentAccess, agents, db, tasks, events, escrow, rateLimiter } = services app.register(async (v1) => { v1.setErrorHandler((/** @type {any} */ err, request, reply) => { @@ -24,6 +57,7 @@ export function registerV1Routes(app, services) { const hint = err.issues.map((/** @type {any} */ issue) => `${(issue.path || []).map((/** @type {any} */ p) => p.key).join('.')}: ${issue.message}`).join('; ') return fail(reply, 400, 'invalid_request', 'Request validation failed', hint) } + if (err.code === 'INSUFFICIENT_BALANCE') return fail(reply, 402, 'insufficient_points', 'Not enough available points to hold in escrow', 'Lower the reward or wait for income.') // Fastify 自己的请求错误(如 JSON 解析失败)带 4xx statusCode;其余都是程序 bug。 if (typeof err.statusCode === 'number' && err.statusCode < 500) return fail(reply, err.statusCode, 'invalid_request', String(err.message), 'Fix the request and do not retry it unchanged.') const traced = /** @type {any} */ (request) @@ -49,6 +83,16 @@ export function registerV1Routes(app, services) { throw new Error(`Unhandled status: ${agent.status}, id: ${agent.id}`) } /** @type {any} */ (request).agent = agent + + // 限流:按 Agent 分读 / 写 / 发任务三类令牌桶。 + const isPublish = request.method === 'POST' && request.routeOptions.url === '/v1/tasks' + for (const endpoint of [request.method === 'GET' ? 'read' : 'write', ...(isPublish ? ['publish'] : [])]) { + const decision = await rateLimiter.consume(request.ctx, { endpoint, subject: agent.id }) + if (!decision.allowed) { + reply.header('Retry-After', String(Math.ceil(decision.retryAfterMs / 1000))) + return fail(reply, 429, 'rate_limited', `Too many ${endpoint} requests`, 'Wait for Retry-After seconds.') + } + } }) v1.post('/agents/register', async (request) => { @@ -65,6 +109,83 @@ export function registerV1Routes(app, services) { await agents.updateProfile(request.ctx, db, { agent_id: agentId, description: body.description, skills: body.skills }) return ok(agentView(await agents.getById(request.ctx, db, { agent_id: agentId }))) }) + + v1.get('/home', async (request) => { + const agent = agentOf(request) + const [wallet, home, unread] = await Promise.all([ + escrow.walletOf(request.ctx, { wallet_user_id: agent.wallet_user_id }), + tasks.homeFor(request.ctx, { agent }), + events.pullUnread(request.ctx, { agent }), + ]) + return ok({ + agent: { id: agent.id, role: agent.role, level: agent.level, xp: agent.xp }, + wallet, + my_active_tasks: home.my_active_tasks.map(taskView), + to_review: home.to_review.map(taskView), + recommended_tasks: home.recommended_tasks.map(taskView), + events: unread.map(eventView), + announcements: [], + skill_version: SKILL_VERSION, + next_heartbeat_hint_sec: home.my_active_tasks.length > 0 || home.to_review.length > 0 ? 1800 : 3600, + }) + }) + + v1.get('/events', async (request) => { + const query = v.parse(EventsQuery, request.query) + return ok({ events: (await events.listSince(request.ctx, { agent: agentOf(request), since: query.since })).map(eventView) }) + }) + + v1.get('/wallet', async (request) => ok(await escrow.walletOf(request.ctx, { wallet_user_id: agentOf(request).wallet_user_id }))) + + v1.get('/tasks', async (request) => { + const query = v.parse(ListQuery, request.query) + const page = await tasks.list(request.ctx, { agent: agentOf(request), ...query }) + return ok({ tasks: page.tasks.map(taskView), next_cursor: page.next_cursor }) + }) + + v1.post('/tasks', async (request, reply) => { + const body = v.parse(PublishBody, request.body) + const key = v.parse(IdempotencyKey, request.headers['idempotency-key']) + const result = await tasks.publish(request.ctx, { agent: agentOf(request), title: body.title, spec: body.spec, template: body.template, reward: body.reward, deadline_at: body.deadline, acceptance: body.acceptance, client_request_key: key }) + if (!result.ok) return refused(reply, result) + return ok({ ...taskView(result.task), acceptance: result.acceptance }) + }) + + v1.get('/tasks/:id', async (request, reply) => { + const params = v.parse(TaskParams, request.params) + const result = await tasks.get(request.ctx, { agent: agentOf(request), task_id: params.id }) + if (!result.ok) return refused(reply, result) + return ok({ ...taskView(result.task), acceptance: result.acceptance, entry_count: result.entry_count }) + }) + + v1.patch('/tasks/:id', async (request, reply) => { + const params = v.parse(TaskParams, request.params) + const body = v.parse(EditBody, request.body) + const key = v.parse(IdempotencyKey, request.headers['idempotency-key']) + return acted(reply, await tasks.act(request.ctx, { agent: agentOf(request), task_id: params.id, action: { type: 'edit', add_reward: body.add_reward ?? 0, deadline_at: body.deadline ?? null }, idempotency_key: key })) + }) + + for (const type of /** @type {const} */ (['cancel', 'claim', 'enter'])) { + v1.post(`/tasks/:id/${type}`, async (request, reply) => { + const params = v.parse(TaskParams, request.params) + return acted(reply, await tasks.act(request.ctx, { agent: agentOf(request), task_id: params.id, action: { type } })) + }) + } + + v1.get('/tasks/:id/comments', async (request, reply) => { + const params = v.parse(TaskParams, request.params) + const result = await tasks.listComments(request.ctx, { agent: agentOf(request), task_id: params.id }) + if (!result.ok) return refused(reply, result) + return ok({ comments: result.comments.map(commentView) }) + }) + + v1.post('/tasks/:id/comments', async (request, reply) => { + const params = v.parse(TaskParams, request.params) + const body = v.parse(CommentBody, request.body) + const result = await tasks.addComment(request.ctx, { agent: agentOf(request), task_id: params.id, body: body.body }) + if (!result.ok) return refused(reply, result) + return ok(commentView(result.comment)) + }) }, { prefix: '/v1' }) } @@ -89,6 +210,69 @@ export function agentView(agent) { } /** + * 任务的对外形状(字段名以接入说明为准:deadline / next_checkpoint_due / review_due)。 + * @param {import('@voko/exchange').Task} task + */ +export function taskView(task) { + return { + id: task.id, + guild_id: task.guild_id, + publisher_agent_id: task.publisher_agent_id, + template: task.template, + title: task.title, + spec: task.spec, + reward: task.reward, + rarity: task.rarity, + status: task.status, + deadline: task.deadline_at.toISOString(), + holder_agent_id: task.holder_agent_id, + next_checkpoint_due: task.next_checkpoint_due_at === null ? null : task.next_checkpoint_due_at.toISOString(), + review_due: task.review_due_at === null ? null : task.review_due_at.toISOString(), + rework_count: task.rework_count, + created_at: task.created_at.toISOString(), + } +} + +/** + * @param {{ id: number, type: string, task_id: string|null, created_at: Date }} event + */ +function eventView(event) { + return { id: event.id, type: event.type, task_id: event.task_id, at: event.created_at.toISOString() } +} + +/** + * @param {{ id: string, author_agent_id: string, body: string, created_at: Date }} comment + */ +function commentView(comment) { + return { id: comment.id, author_agent_id: comment.author_agent_id, body: comment.body, created_at: comment.created_at.toISOString() } +} + +/** + * 鉴权钩子挂在请求上的 Agent。 + * @param {import('fastify').FastifyRequest} request + * @returns {import('@voko/exchange').Agent} + */ +function agentOf(request) { + return /** @type {any} */ (request).agent +} + +/** + * @param {import('fastify').FastifyReply} reply + * @param {{ ok: true, task: import('@voko/exchange').Task } | { ok: false, code: keyof typeof REFUSAL_STATUS, hint: string }} result + */ +function acted(reply, result) { + return result.ok ? ok(taskView(result.task)) : refused(reply, result) +} + +/** + * @param {import('fastify').FastifyReply} reply + * @param {{ code: keyof typeof REFUSAL_STATUS, hint: string }} refusal + */ +function refused(reply, refusal) { + return fail(reply, REFUSAL_STATUS[refusal.code], refusal.code, refusal.hint, refusal.hint) +} + +/** * @param {unknown} data */ function ok(data) { diff --git a/apps/voko-server/src/voko-worker-main.js b/apps/voko-server/src/voko-worker-main.js index 6d99002..bd6bf85 100644 --- a/apps/voko-server/src/voko-worker-main.js +++ b/apps/voko-server/src/voko-worker-main.js @@ -1,5 +1,5 @@ /** - * `voko-worker` 进程入口:领取并执行 vk_jobs(W2 起实现)。骨架阶段只做心跳,证明进程可被 PM2 托管。 + * `voko-worker` 进程入口:每 2 秒领取并执行到期的 vk_jobs(见 SDD「定时任务与事件」)。 */ import { createRuntime } from './assembly/runtime.js' @@ -12,8 +12,7 @@ let ticks = 0 let stopping = false async function tick() { - // W2:在这里用 FOR UPDATE SKIP LOCKED 领取到期 job(见 SDD「定时任务与事件」)。 - await runtime.services.db.raw('select 1') + await runtime.services.jobs.runDue({}) ticks += 1 if (ticks % HEARTBEAT_EVERY_TICKS === 0) runtime.logger.info('worker 心跳', { ticks }) } diff --git a/apps/voko-server/tools/scenarios/voko-alpha/_lib.mjs b/apps/voko-server/tools/scenarios/voko-alpha/_lib.mjs new file mode 100644 index 0000000..3a982c8 --- /dev/null +++ b/apps/voko-server/tools/scenarios/voko-alpha/_lib.mjs @@ -0,0 +1,185 @@ +/** + * 场景脚本公共库:HTTP 客户端、断言、通过真实接口建主人 / 公会 / Agent、对账、按场景清理。 + * + * 前提:voko-server 已启动(VOKO_API,默认 http://localhost:9201),.env 已配置真实 SMTP; + * VOKO_TEST_EMAIL、VOKO_ADMIN_EMAIL / VOKO_ADMIN_PASSWORD。缺任一项输出 BLOCKED 并以 2 退出。 + */ +import { randomUUID } from 'node:crypto' +import { createBillingRepo, createLedger } from '@drama/billing' +import { createDb } from '@drama/db' + +import { databaseUrlFromArgs } from '../../../src/db/env-file.js' +import { verifyConservation } from './verify-conservation.mjs' + +const api = (process.env.VOKO_API || 'http://localhost:9201').replace(/\/$/, '') + +/** + * @param {string} name 场景名,用于测试邮箱与公会 slug + * @param {(world: ReturnType) => Promise} body + */ +export async function runScenario(name, body) { + const missing = ['VOKO_TEST_EMAIL', 'VOKO_ADMIN_EMAIL', 'VOKO_ADMIN_PASSWORD'].filter((key) => !process.env[key]) + if (missing.length > 0) { + console.log(JSON.stringify({ status: 'BLOCKED', missing })) + process.exit(2) + } + const db = createDb({ url: databaseUrlFromArgs(process.argv.slice(2)), pool: { min: 1, max: 2 } }) + const world = createWorld(name, db) + try { + await body(world) + const problems = await verifyConservation(db) + world.check('对账:托管与账本守恒', problems.length === 0, problems) + console.log(JSON.stringify({ status: 'PASS', scenario: name, results: world.results }, null, 2)) + } catch (err) { + console.log(JSON.stringify({ status: 'FAIL', scenario: name, results: world.results, error: err instanceof Error ? err.message : String(err) }, null, 2)) + process.exitCode = 1 + } finally { + await world.cleanup() + await db.destroy() + } +} + +/** + * @param {string} name + * @param {import('knex').Knex} db + */ +function createWorld(name, db) { + const run = Date.now() + const [local, domain] = String(process.env.VOKO_TEST_EMAIL).split('@') + const results = [] + const emails = [] + const userIds = [] + const guildIds = [] + const agentIds = [] + let adminToken = null + + /** + * @param {string} label + * @param {boolean} pass + * @param {unknown} [detail] + */ + function check(label, pass, detail) { + results.push({ name: label, pass, ...(pass ? {} : { detail }) }) + if (!pass) throw new Error(`FAILED: ${label} ${JSON.stringify(detail)}`) + } + + /** + * @param {string} method + * @param {string} path + * @param {{ body?: unknown, token?: string, idempotency_key?: string }} [opts] + */ + async function call(method, path, opts = {}) { + const response = await fetch(`${api}${path}`, { + method, + headers: { + ...(opts.body === undefined ? {} : { 'content-type': 'application/json' }), + ...(opts.token === undefined ? {} : { authorization: `Bearer ${opts.token}` }), + ...(opts.idempotency_key === undefined ? {} : { 'idempotency-key': opts.idempotency_key }), + }, + body: opts.body === undefined ? undefined : JSON.stringify(opts.body), + }) + return { status: response.status, json: await response.json() } + } + + /** 等一封真实投递的邮件落库(发信不阻塞接口返回),最多 30 秒。 */ + async function waitForEmail(where) { + for (let i = 0; i < 30; i += 1) { + const log = await db('email_delivery_logs').where(where).orderBy('created_at', 'desc').first() + if (log !== undefined) return log + await new Promise((resolve) => setTimeout(resolve, 1000)) + } + throw new Error(`email not delivered within 30s: ${JSON.stringify(where)}`) + } + + async function admin() { + if (adminToken === null) { + const login = await call('POST', '/api/voko/auth/login', { body: { method: 'password', email: process.env.VOKO_ADMIN_EMAIL, password: process.env.VOKO_ADMIN_PASSWORD } }) + check('admin login', login.status === 200, login) + adminToken = login.json.access_token + } + return adminToken + } + + /** 一个已通过加入申请的主人:真实验证码注册 → 提交申请 → 运营通过。 */ + async function createOwner(tag) { + const email = `${local}+voko-${name}-${tag}-${run}@${domain}`.toLowerCase() + emails.push(email) + const sent = await call('POST', '/api/voko/auth/send-code', { body: { email, locale: 'en' } }) + check(`主人 ${tag}:send-code`, sent.status === 200, sent) + const codeLog = await waitForEmail({ to_email: email, template: 'verify-code', status: 'sent' }) + const registered = await call('POST', '/api/voko/auth/register', { body: { email, password: `pw-${randomUUID()}`, verification_code: JSON.parse(codeLog.metadata_json).code, locale: 'en' } }) + check(`主人 ${tag}:注册`, registered.status === 200, registered) + const owner = { user_id: registered.json.user.id, token: registered.json.access_token, email } + userIds.push(owner.user_id) + const request = await call('POST', '/api/voko/access-requests', { token: owner.token, body: { note: `${name} scenario` } }) + const approved = await call('POST', `/api/voko/admin/access-requests/${request.json.access_request.id}/review`, { token: await admin(), body: { decision: 'approved' } }) + check(`主人 ${tag}:申请通过`, approved.status === 200, approved) + return owner + } + + async function createGuild(tag, leadUserId) { + const guild = await call('POST', '/api/voko/admin/guilds', { token: await admin(), body: { name: `${name} ${tag}`, slug: `${name}-${tag}-${run}`, lead_user_id: leadUserId, require_lead_approval: false } }) + check(`建公会 ${tag}`, guild.status === 200, guild) + guildIds.push(guild.json.guild.id) + return guild.json.guild.id + } + + /** 注册一个 Agent 并由主人认领进公会;返回 { id, key }。 */ + async function createAgent(tag, owner, guildId) { + const registered = await call('POST', '/v1/agents/register', { body: { name: `${name}-${tag}-${run}`, skills: ['writing'] } }) + check(`Agent ${tag}:注册`, registered.status === 200, registered) + const data = registered.json.data + agentIds.push(data.agent_id) + const claimed = await call('POST', `/api/voko/claims/${data.claim_url.split('/claim/')[1]}`, { token: owner.token, body: { verification_code: data.verification_code, guild_id: guildId } }) + check(`Agent ${tag}:认领`, claimed.status === 200 && claimed.json.agent.status === 'active', claimed) + return { id: data.agent_id, key: data.api_key } + } + + /** 运营给 Agent 钱包调账(公司预算走人工调账,reason=admin_adjustment)。 */ + async function grantBudget(agentId, amount) { + const agent = await db('vk_agents').where({ id: agentId }).first('wallet_user_id') + await createLedger(createBillingRepo(db)).grant({ user_id: agent.wallet_user_id, amount, source: 'admin', reason: 'admin_adjustment', note: `${name} scenario budget`, idempotency_key: `scenario:${name}:${run}:${agentId}:${amount}` }) + } + + /** 发布一个任务,返回接口响应。 */ + function publish(agent, overrides = {}, idempotencyKey = randomUUID()) { + return call('POST', '/v1/tasks', { + token: agent.key, + idempotency_key: idempotencyKey, + body: { title: 'Translate onboarding email', spec: 'Translate the onboarding email to Japanese.', template: 'piecework', reward: 10, deadline: new Date(Date.now() + 86400_000).toISOString(), acceptance: [{ id: 'c1', text: 'Natural Japanese' }], ...overrides }, + }) + } + + async function cleanup() { + const taskIds = guildIds.length === 0 ? [] : (await db('vk_tasks').whereIn('guild_id', guildIds).select('id')).map((row) => row.id) + const wallets = agentIds.length === 0 ? [] : (await db('vk_agents').whereIn('id', agentIds).select('wallet_user_id')).map((row) => row.wallet_user_id) + if (taskIds.length > 0) { + for (const table of ['vk_task_acceptance_items', 'vk_task_holds', 'vk_task_entries', 'vk_comments', 'vk_escrow_payouts', 'vk_escrows']) await db(table).whereIn('task_id', taskIds).del() + await db('vk_jobs').whereIn('ref_id', taskIds).del() + await db('vk_tasks').whereIn('id', taskIds).del() + } + if (guildIds.length > 0) { + await db('vk_events').whereIn('guild_id', guildIds).del() + await db('vk_guild_members').whereIn('guild_id', guildIds).del() + await db('vk_guilds').whereIn('id', guildIds).del() + } + if (agentIds.length > 0) { + await db('billing_ledger').whereIn('user_id', wallets).del() + await db('vk_api_keys').whereIn('agent_id', agentIds).del() + await db('vk_agent_claims').whereIn('agent_id', agentIds).del() + await db('vk_agents').whereIn('id', agentIds).del() + await db('users').whereIn('id', wallets).del() + } + if (userIds.length > 0) { + await db('vk_access_requests').whereIn('user_id', userIds).del() + await db('auth_sessions').whereIn('user_id', userIds).del() + await db('users').whereIn('id', userIds).del() + } + if (emails.length > 0) { + await db('auth_codes').whereIn('email', emails).del() + await db('email_delivery_logs').whereIn('to_email', emails).del() + } + } + + return { db, results, check, call, createOwner, createGuild, createAgent, grantBudget, publish, cleanup } +} diff --git a/apps/voko-server/tools/scenarios/voko-alpha/concurrency.mjs b/apps/voko-server/tools/scenarios/voko-alpha/concurrency.mjs new file mode 100644 index 0000000..5424e19 --- /dev/null +++ b/apps/voko-server/tools/scenarios/voko-alpha/concurrency.mjs @@ -0,0 +1,35 @@ +#!/usr/bin/env node +/** + * 并发与幂等的真实链路脚本(Planning 用例 CC-01、CC-02;CC-03 结算并发在 W4)。 + * CC-01 用 6 个 Agent(每个主人最多 3 个)同时发 20 个认领请求,断言恰好 1 个成功。 + */ +import { randomUUID } from 'node:crypto' + +import { runScenario } from './_lib.mjs' + +await runScenario('cc', async ({ db, check, call, createOwner, createGuild, createAgent, publish }) => { + const alice = await createOwner('alice') + const bob = await createOwner('bob') + const carol = await createOwner('carol') + const guild = await createGuild('main', alice.user_id) + const publisher = await createAgent('publisher', alice, guild) + const takers = [] + for (const [owner, tag] of [[bob, 'bob'], [carol, 'carol']]) { + for (let i = 1; i <= 3; i += 1) takers.push(await createAgent(`${tag}-${i}`, owner, guild)) + } + + // CC-02 同一 Idempotency-Key 并发发布 10 次 → 1 个任务、1 笔扣款。 + const key = randomUUID() + const replays = await Promise.all(Array.from({ length: 10 }, () => publish(publisher, { reward: 30 }, key))) + const taskIds = new Set(replays.map((r) => r.json.data && r.json.data.id)) + const wallet = (await call('GET', '/v1/wallet', { token: publisher.key })).json.data + const holds = await db('billing_ledger').where({ reason: 'vk_escrow_hold', ref_id: [...taskIds][0] }) + check('CC-02 发布重放 10 次 → 1 个任务、1 笔扣款', replays.every((r) => r.status === 200) && taskIds.size === 1 && holds.length === 1 && wallet.available === 70 && wallet.held === 30, { statuses: replays.map((r) => r.status), task_ids: [...taskIds], wallet }) + + // CC-01 20 个并发认领同一计件活 → 恰好 1 个成功。 + const taskId = [...taskIds][0] + const claims = await Promise.all(Array.from({ length: 20 }, (_, i) => call('POST', `/v1/tasks/${taskId}/claim`, { token: takers[i % takers.length].key }))) + const won = claims.filter((r) => r.status === 200) + const openHolds = await db('vk_task_holds').where({ task_id: taskId }).whereNull('ended_at') + check('CC-01 20 个并发认领 → 恰好 1 个成功,其余 409', won.length === 1 && claims.filter((r) => r.status === 409).length === 19 && openHolds.length === 1 && openHolds[0].agent_id === won[0].json.data.holder_agent_id, { statuses: claims.map((r) => r.status) }) +}) diff --git a/apps/voko-server/tools/scenarios/voko-alpha/f1-publish.mjs b/apps/voko-server/tools/scenarios/voko-alpha/f1-publish.mjs new file mode 100644 index 0000000..27df057 --- /dev/null +++ b/apps/voko-server/tools/scenarios/voko-alpha/f1-publish.mjs @@ -0,0 +1,84 @@ +#!/usr/bin/env node +/** + * F1 发布的真实链路脚本(Planning 用例 F1-01、02、03、05、06、07、08;F1-04 发布前审核在 W3)。 + */ +import { randomUUID } from 'node:crypto' + +import { runScenario } from './_lib.mjs' + +await runScenario('f1', async ({ db, check, call, createOwner, createGuild, createAgent, publish }) => { + const alice = await createOwner('alice') + const bob = await createOwner('bob') + const guild = await createGuild('main', alice.user_id) + const otherGuild = await createGuild('other', bob.user_id) + const publisher = await createAgent('publisher', alice, guild) + const member = await createAgent('member', bob, guild) + const outsider = await createAgent('outsider', bob, otherGuild) + const wallet = async (agent) => (await call('GET', '/v1/wallet', { token: agent.key })).json.data + const taskCount = async () => Number((await db('vk_tasks').where({ publisher_agent_id: publisher.id }).count({ n: '*' }).first()).n) + + // F1-01 入口校验与余额不足。 + const noList = await publish(publisher, { acceptance: [] }) + check('F1-01 缺验收清单 → 400', noList.status === 400 && noList.json.error.code === 'invalid_request', noList) + const sixItems = await publish(publisher, { acceptance: [1, 2, 3, 4, 5, 6].map((n) => ({ id: `c${n}`, text: `item ${n}` })) }) + check('F1-01 清单 6 条 → 400', sixItems.status === 400, sixItems) + const noKey = await call('POST', '/v1/tasks', { token: publisher.key, body: { title: 't', spec: 's', template: 'piecework', reward: 10, deadline: new Date(Date.now() + 3600_000).toISOString(), acceptance: [{ id: 'c1', text: 'x' }] } }) + check('缺 Idempotency-Key → 400', noKey.status === 400, noKey) + const tooRich = await publish(publisher, { reward: 101 }) + const ledgerAfterRefusal = await db('billing_ledger').where({ source: 'voko', reason: 'vk_escrow_hold' }).whereIn('user_id', db('vk_agents').where({ id: publisher.id }).select('wallet_user_id')) + check('F1-01 余额不足 → 402,无任务、无流水', tooRich.status === 402 && tooRich.json.error.code === 'insufficient_points' && (await taskCount()) === 0 && ledgerAfterRefusal.length === 0, tooRich) + + // F1-02 / F1-05 员工 Agent 用自己的积分发布:余额立即减少、托管 held、稀有度按区间。 + const published = await publish(publisher, { reward: 60 }) + const task = published.json.data + const escrow = await db('vk_escrows').where({ task_id: task.id }).first() + check('F1-02 发布成功 → open、稀有度 2', published.status === 200 && task.status === 'open' && task.rarity === 2 && task.acceptance.length === 1, published) + check('F1-02 余额立即减 reward、托管 held', JSON.stringify(await wallet(publisher)) === JSON.stringify({ available: 40, held: 60 }) && escrow.status === 'held' && escrow.amount === 60, { wallet: await wallet(publisher), escrow }) + + // F1-03 非成员看不到;成员下一次 /home 能看到。 + const outsiderGet = await call('GET', `/v1/tasks/${task.id}`, { token: outsider.key }) + const outsiderList = await call('GET', '/v1/tasks', { token: outsider.key }) + check('F1-03 非成员看不到任务', outsiderGet.status === 403 && outsiderList.json.data.tasks.every((t) => t.id !== task.id), { outsiderGet, outsiderList }) + const home = await call('GET', '/v1/home', { token: member.key }) + check('F1-03 成员 /home 推荐里有新任务与 task.published 事件', home.status === 200 && home.json.data.recommended_tasks.some((t) => t.id === task.id) && home.json.data.events.some((e) => e.type === 'task.published' && e.task_id === task.id), home) + const homeAgain = await call('GET', '/v1/home', { token: member.key }) + const eventsSince = await call('GET', '/v1/events?since=0', { token: member.key }) + check('事件游标:/home 已读后不再返回,/events?since=0 仍可补取', homeAgain.json.data.events.length === 0 && eventsSince.json.data.events.some((e) => e.task_id === task.id), { homeAgain, eventsSince }) + const memberList = await call('GET', '/v1/tasks?status=open&template=piecework', { token: member.key }) + check('成员看板按条件筛选能看到', memberList.json.data.tasks.some((t) => t.id === task.id), memberList) + + // F1-08 评论区。 + const comment = await call('POST', `/v1/tasks/${task.id}/comments`, { token: member.key, body: { body: 'Which tone should the translation use?' } }) + const outsiderComment = await call('POST', `/v1/tasks/${task.id}/comments`, { token: outsider.key, body: { body: 'hi' } }) + const comments = await call('GET', `/v1/tasks/${task.id}/comments`, { token: publisher.key }) + const publisherEvents = await call('GET', '/v1/events?since=0', { token: publisher.key }) + check('F1-08 成员可评论、非成员 403', comment.status === 200 && outsiderComment.status === 403 && comments.json.data.comments.length === 1, { comment, outsiderComment }) + check('F1-08 评论进入发布方的事件', publisherEvents.json.data.events.some((e) => e.type === 'comment.created' && e.task_id === task.id), publisherEvents) + + // F1-06 open 时加赏金、延长截止;重放不重复扣;被认领后修改 → 409。 + const editKey = randomUUID() + const later = new Date(Date.now() + 2 * 86400_000).toISOString() + const edited = await call('PATCH', `/v1/tasks/${task.id}`, { token: publisher.key, idempotency_key: editKey, body: { add_reward: 20, deadline: later } }) + check('F1-06 加赏金(补扣托管)与延长截止', edited.status === 200 && edited.json.data.reward === 80 && edited.json.data.deadline === later && JSON.stringify(await wallet(publisher)) === JSON.stringify({ available: 20, held: 80 }), { edited, wallet: await wallet(publisher) }) + const editReplay = await call('PATCH', `/v1/tasks/${task.id}`, { token: publisher.key, idempotency_key: editKey, body: { add_reward: 20, deadline: later } }) + check('F1-06 同一 Idempotency-Key 重放不重复补扣', editReplay.status === 200 && editReplay.json.data.reward === 80 && (await wallet(publisher)).available === 20, editReplay) + const shorten = await call('PATCH', `/v1/tasks/${task.id}`, { token: publisher.key, idempotency_key: randomUUID(), body: { deadline: new Date(Date.now() + 3600_000).toISOString() } }) + check('F1-06 缩短截止 → 422', shorten.status === 422 && shorten.json.error.code === 'rule_violation', shorten) + const memberEdit = await call('PATCH', `/v1/tasks/${task.id}`, { token: member.key, idempotency_key: randomUUID(), body: { add_reward: 1 } }) + check('非发布方修改 → 403', memberEdit.status === 403, memberEdit) + const claimed = await call('POST', `/v1/tasks/${task.id}/claim`, { token: member.key }) + check('成员认领 → claimed', claimed.status === 200 && claimed.json.data.status === 'claimed' && claimed.json.data.holder_agent_id === member.id, claimed) + const editClaimed = await call('PATCH', `/v1/tasks/${task.id}`, { token: publisher.key, idempotency_key: randomUUID(), body: { add_reward: 1 } }) + const cancelClaimed = await call('POST', `/v1/tasks/${task.id}/cancel`, { token: publisher.key }) + check('F1-06 / F1-07 被认领后修改或撤回 → 409', editClaimed.status === 409 && cancelClaimed.status === 409, { editClaimed, cancelClaimed }) + + // F1-07 无人报名撤回全额退款;擂台赛有人报名后撤回 → 409。 + const second = (await publish(publisher, { reward: 15 })).json.data + const cancelled = await call('POST', `/v1/tasks/${second.id}/cancel`, { token: publisher.key }) + const refundedEscrow = await db('vk_escrows').where({ task_id: second.id }).first() + check('F1-07 无人报名撤回 → cancelled、全额退款', cancelled.status === 200 && cancelled.json.data.status === 'cancelled' && refundedEscrow.status === 'refunded' && JSON.stringify(await wallet(publisher)) === JSON.stringify({ available: 20, held: 80 }), { cancelled, wallet: await wallet(publisher) }) + const contest = (await publish(publisher, { reward: 10, template: 'contest' })).json.data + const entered = await call('POST', `/v1/tasks/${contest.id}/enter`, { token: member.key }) + const cancelContest = await call('POST', `/v1/tasks/${contest.id}/cancel`, { token: publisher.key }) + check('F1-07 有人报名后撤回 → 409', entered.status === 200 && entered.json.data.status === 'in_contest' && cancelContest.status === 409, { entered, cancelContest }) +}) diff --git a/apps/voko-server/tools/scenarios/voko-alpha/f2-execute.mjs b/apps/voko-server/tools/scenarios/voko-alpha/f2-execute.mjs new file mode 100644 index 0000000..c5e0a8a --- /dev/null +++ b/apps/voko-server/tools/scenarios/voko-alpha/f2-execute.mjs @@ -0,0 +1,50 @@ +#!/usr/bin/env node +/** + * F2 接单的真实链路脚本(Planning 用例 F2-01、F2-02 的报名部分;检查点、接力、转让、提交在 W3)。 + */ +import { runScenario } from './_lib.mjs' + +await runScenario('f2', async ({ check, call, createOwner, createGuild, createAgent, grantBudget, publish }) => { + const alice = await createOwner('alice') + const bob = await createOwner('bob') + const guild = await createGuild('main', alice.user_id) + const publisher = await createAgent('publisher', alice, guild) + const sibling = await createAgent('sibling', alice, guild) + const taker = await createAgent('taker', bob, guild) + const rival = await createAgent('rival', bob, guild) + await grantBudget(publisher.id, 1000) + + // F2-01 自接自单、同主人、并发上限、稀有度。 + const tasks = [] + for (let i = 0; i < 4; i += 1) tasks.push((await publish(publisher, { reward: 10 })).json.data) + const own = await call('POST', `/v1/tasks/${tasks[0].id}/claim`, { token: publisher.key }) + const sameOwner = await call('POST', `/v1/tasks/${tasks[0].id}/claim`, { token: sibling.key }) + check('F2-01 认领自己或同主人 Agent 的任务 → 409', own.status === 409 && sameOwner.status === 409 && own.json.error.code === 'state_conflict', { own, sameOwner }) + for (let i = 0; i < 3; i += 1) { + const claimed = await call('POST', `/v1/tasks/${tasks[i].id}/claim`, { token: taker.key }) + check(`第 ${i + 1} 个并发持有成功`, claimed.status === 200 && typeof claimed.json.data.next_checkpoint_due === 'string', claimed) + } + const fourth = await call('POST', `/v1/tasks/${tasks[3].id}/claim`, { token: taker.key }) + check('F2-01 第 4 个并发持有 → 409', fourth.status === 409, fourth) + const taken = await call('POST', `/v1/tasks/${tasks[0].id}/claim`, { token: rival.key }) + check('已被认领的任务再认领 → 409', taken.status === 409, taken) + const rare = (await publish(publisher, { reward: 250 })).json.data + const tooRare = await call('POST', `/v1/tasks/${rare.id}/claim`, { token: rival.key }) + check('F2-01 稀有度超过等级上限 → 422', rare.rarity === 4 && tooRare.status === 422 && tooRare.json.error.code === 'rule_violation', { rare, tooRare }) + const home = await call('GET', '/v1/home', { token: taker.key }) + check('/home:手上 3 个任务、心跳 1800、推荐里没有超稀有度任务', home.json.data.my_active_tasks.length === 3 && home.json.data.next_heartbeat_hint_sec === 1800 && home.json.data.recommended_tasks.every((t) => t.id !== rare.id), home) + const publisherEvents = await call('GET', '/v1/events?since=0', { token: publisher.key }) + check('认领通知发布方(task.claimed)', publisherEvents.json.data.events.filter((e) => e.type === 'task.claimed').length === 3, publisherEvents) + + // F2-02 擂台赛多人报名;重复报名、认领擂台赛被拒。 + const contest = (await publish(publisher, { reward: 20, template: 'contest' })).json.data + const claimContest = await call('POST', `/v1/tasks/${contest.id}/claim`, { token: taker.key }) + check('擂台赛不能 claim → 422', claimContest.status === 422, claimContest) + const first = await call('POST', `/v1/tasks/${contest.id}/enter`, { token: taker.key }) + const second = await call('POST', `/v1/tasks/${contest.id}/enter`, { token: rival.key }) + const detail = await call('GET', `/v1/tasks/${contest.id}`, { token: publisher.key }) + check('F2-02 擂台赛多人报名均成功', first.status === 200 && second.status === 200 && detail.json.data.status === 'in_contest' && detail.json.data.entry_count === 2, { first, second, detail }) + const again = await call('POST', `/v1/tasks/${contest.id}/enter`, { token: taker.key }) + const ownContest = await call('POST', `/v1/tasks/${contest.id}/enter`, { token: sibling.key }) + check('重复报名、同主人报名 → 409', again.status === 409 && ownContest.status === 409, { again, ownContest }) +}) diff --git a/apps/voko-server/tools/scenarios/voko-alpha/timers.mjs b/apps/voko-server/tools/scenarios/voko-alpha/timers.mjs new file mode 100644 index 0000000..9d87178 --- /dev/null +++ b/apps/voko-server/tools/scenarios/voko-alpha/timers.mjs @@ -0,0 +1,22 @@ +#!/usr/bin/env node +/** + * 时间类规则的真实链路脚本,由真实 voko-worker 触发(运行前须已启动 `pnpm --filter @voko/server worker`)。 + * 目前覆盖 TM-05(open 过截止无人接 → cancelled 退款);TM-01–04、TM-06、F0-06 在 W3–W4 加入。 + */ +import { runScenario } from './_lib.mjs' + +await runScenario('tm', async ({ check, call, createOwner, createGuild, createAgent, publish }) => { + const alice = await createOwner('alice') + const guild = await createGuild('main', alice.user_id) + const publisher = await createAgent('publisher', alice, guild) + + const published = await publish(publisher, { reward: 40, deadline: new Date(Date.now() + 4000).toISOString() }) + check('发布 4 秒后截止的任务', published.status === 200 && published.json.data.status === 'open', published) + let task = published.json.data + for (let i = 0; i < 30 && task.status === 'open'; i += 1) { + await new Promise((resolve) => setTimeout(resolve, 1000)) + task = (await call('GET', `/v1/tasks/${task.id}`, { token: publisher.key })).json.data + } + const wallet = (await call('GET', '/v1/wallet', { token: publisher.key })).json.data + check('TM-05 open 过截止 → worker 取消并全额退款', task.status === 'cancelled' && wallet.available === 100 && wallet.held === 0, { task, wallet }) +}) diff --git a/apps/voko-server/tools/scenarios/voko-alpha/verify-conservation.mjs b/apps/voko-server/tools/scenarios/voko-alpha/verify-conservation.mjs new file mode 100644 index 0000000..d390a6d --- /dev/null +++ b/apps/voko-server/tools/scenarios/voko-alpha/verify-conservation.mjs @@ -0,0 +1,52 @@ +#!/usr/bin/env node +/** + * 对账(SDD「积分清算」):① 已关闭托管的发放明细之和 = 托管金额;② held 托管都对应未结束的任务, + * 未结束的任务都有 held 托管;③ 每笔 voko 流水的 ref 都能找到业务记录。 + * 任一不满足即失败并打印数据 id。可单独运行,也在每个场景脚本末尾运行。 + */ +import { fileURLToPath } from 'node:url' +import { createDb } from '@drama/db' + +import { databaseUrlFromArgs } from '../../../src/db/env-file.js' + +const TERMINAL = ['settled', 'cancelled', 'refunded'] + +/** + * @param {import('knex').Knex} db + * @returns {Promise>} + */ +export async function verifyConservation(db) { + const problems = [] + const closed = await db('vk_escrows as e') + .leftJoin('vk_escrow_payouts as p', 'p.escrow_id', 'e.id') + .whereNot('e.status', 'held') + .groupBy('e.id', 'e.amount') + .havingRaw('COALESCE(SUM(p.amount), 0) <> e.amount') + .select('e.id', 'e.amount', db.raw('COALESCE(SUM(p.amount), 0) as paid')) + for (const row of closed) problems.push({ rule: 'payouts_sum', id: row.id, detail: `amount=${row.amount}, payouts=${row.paid}` }) + + const heldOnEnded = await db('vk_escrows as e').join('vk_tasks as t', 't.id', 'e.task_id').where('e.status', 'held').whereIn('t.status', TERMINAL).select('e.id', 't.status') + for (const row of heldOnEnded) problems.push({ rule: 'held_on_ended_task', id: row.id, detail: `task status=${row.status}` }) + const openWithoutHeld = await db('vk_tasks as t') + .leftJoin('vk_escrows as e', (join) => join.on('e.task_id', 't.id').andOnVal('e.status', 'held')) + .whereNotIn('t.status', TERMINAL) + .whereNull('e.id') + .select('t.id', 't.status') + for (const row of openWithoutHeld) problems.push({ rule: 'active_task_without_held_escrow', id: row.id, detail: `task status=${row.status}` }) + + const orphanTaskRefs = await db('billing_ledger as l').leftJoin('vk_tasks as t', 't.id', 'l.ref_id').where({ 'l.source': 'voko', 'l.ref_type': 'vk_task' }).whereNull('t.id').select('l.id', 'l.ref_id') + const orphanAgentRefs = await db('billing_ledger as l').leftJoin('vk_agents as a', 'a.id', 'l.ref_id').where({ 'l.source': 'voko', 'l.ref_type': 'vk_agent' }).whereNull('a.id').select('l.id', 'l.ref_id') + for (const row of [...orphanTaskRefs, ...orphanAgentRefs]) problems.push({ rule: 'ledger_ref_missing', id: row.id, detail: `ref_id=${row.ref_id}` }) + return problems +} + +if (process.argv[1] === fileURLToPath(import.meta.url)) { + const db = createDb({ url: databaseUrlFromArgs(process.argv.slice(2)), pool: { min: 1, max: 1 } }) + try { + const problems = await verifyConservation(db) + console.log(JSON.stringify({ status: problems.length === 0 ? 'PASS' : 'FAIL', problems }, null, 2)) + if (problems.length > 0) process.exitCode = 1 + } finally { + await db.destroy() + } +} diff --git a/docs/runbooks/README.md b/docs/runbooks/README.md index 9e82423..5fdbe39 100644 --- a/docs/runbooks/README.md +++ b/docs/runbooks/README.md @@ -8,6 +8,7 @@ - 初始化、升级或检查数据库结构:[数据库结构与迁移](../database.md)。 - 验收仓库骨架(注册 → 申请 → 审核 → 记账幂等):[骨架真实链路验收](verify-skeleton.md);本地一键:`VOKO_TEST_EMAIL=… bash tools/scripts/verify-local.sh`。 - 验收 F0 入驻(Agent 注册 → 认领 → 启动积分 → 3 个上限):[F0 入驻真实链路验收](verify-f0.md)。 +- 验收任务发布、认领、并发幂等、定时规则与对账:[VOKO Alpha 验收脚本](verify-alpha.md)。 ## 部署 diff --git a/docs/runbooks/verify-alpha.md b/docs/runbooks/verify-alpha.md new file mode 100644 index 0000000..3ef96de --- /dev/null +++ b/docs/runbooks/verify-alpha.md @@ -0,0 +1,33 @@ +# VOKO Alpha 验收脚本 + +按 Planning 的用例编号验收任务交换的真实链路:真实 HTTP、真实 MySQL、真实 SMTP、真实 worker。每个脚本自己建主人、公会与 Agent,末尾跑一次对账,结束时只清理自己写入的数据。 + +## 前提 + +- 本地库已 `pnpm --filter @voko/server db:migrate` 到最新;voko-server 已启动;跑 `timers` 时 voko-worker 也要在跑(`pnpm --filter @voko/server worker`)。 +- `apps/voko-server/.env` 配置了真实 SMTP;`VOKO_TEST_EMAIL` 是能收信的邮箱;有一个管理员账号(本地 `pnpm dev:user admin@example.test admin123`)。 +- 每个脚本会给测试邮箱的 plus 地址发若干封验证码与审核通过邮件。 + +## 执行 + +```bash +export VOKO_TEST_EMAIL=you@gmail.com VOKO_ADMIN_EMAIL=admin@example.test VOKO_ADMIN_PASSWORD=admin123 +pnpm --filter @voko/server verify:f0 # F0-01–05 入驻,见 verify-f0.md +pnpm --filter @voko/server verify:f1 # F1-01、02、03、05–08 发布、修改、撤回、评论 +pnpm --filter @voko/server verify:f2 # F2-01、F2-02 认领规则与擂台赛报名 +pnpm --filter @voko/server verify:concurrency # CC-01 并发认领、CC-02 发布幂等 +pnpm --filter @voko/server verify:timers # TM-05 过期退款(需 worker) +pnpm --filter @voko/server verify:conservation # 只跑对账 +``` + +本地一键:`VOKO_TEST_EMAIL=… bash tools/scripts/verify-local.sh` 会起库、API 与 worker 并依次跑以上全部。 + +## 完成判据 + +- 每个脚本输出 `"status": "PASS"`,其中最后一项「对账:托管与账本守恒」通过。 +- 缺少环境变量输出 `BLOCKED` 并以 2 退出,不算通过。 +- 失败时 `detail` 里带接口响应或数据 id;服务端错误看 `tmp-voko-server.log`、`tmp-voko-worker.log`。 + +## 尚未覆盖 + +F1-04(发布前审核)、F2-03–07、F3、TM-01–04 在 W3;F0-06、TM-06、CC-03、WH、E2E 在 W4。CC-01 用 6 个 Agent 发 20 个并发请求(每个主人最多认领 3 个 Agent)。 diff --git a/packages/voko-clearing/package.json b/packages/voko-clearing/package.json index 64dceb4..7da9fb1 100644 --- a/packages/voko-clearing/package.json +++ b/packages/voko-clearing/package.json @@ -5,5 +5,8 @@ "type": "module", "exports": { ".": "./src/index.js" + }, + "dependencies": { + "@drama/billing": "workspace:*" } } diff --git a/packages/voko-clearing/src/escrow.js b/packages/voko-clearing/src/escrow.js new file mode 100644 index 0000000..d9b3485 --- /dev/null +++ b/packages/voko-clearing/src/escrow.js @@ -0,0 +1,112 @@ +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' + +/** + * 托管:发布时从发布方钱包扣款并记 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 }) + } + + /** + * 钱包视图:可用 = 账本余额;冻结 = 该钱包名下 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) } + } + + return { hold, topupExists, topup, refund, walletOf } +} + +/** + * 取任务的 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}` +} diff --git a/packages/voko-clearing/src/index.js b/packages/voko-clearing/src/index.js index 85a9a31..2780ae3 100644 --- a/packages/voko-clearing/src/index.js +++ b/packages/voko-clearing/src/index.js @@ -1,9 +1,10 @@ /** * `@voko/clearing` —— VOKO 的积分清算:托管与发放明细;动账一律经 `@drama/billing` 的账本。 * - * 当前只有表结构;托管、结算、退款 service 在 W2–W4 加入。 + * 当前有托管的冻结、补扣与退回;结算(含接力分配)、打赏与对账在 W4 加入。 * * 边界:不读 `process.env`、不建连接;db 由 composition root 注入。 */ +export { createEscrow } from './escrow.js' export { default as clearingDbSchema } from './db/tables/index.cjs' diff --git a/packages/voko-exchange/src/events.js b/packages/voko-exchange/src/events.js new file mode 100644 index 0000000..fc2ad86 --- /dev/null +++ b/packages/voko-exchange/src/events.js @@ -0,0 +1,53 @@ +const EVENTS = 'vk_events' +const PAGE = 50 + +/** + * @typedef {{ id: number, type: string, task_id: string|null, created_at: Date }} AgentEvent + */ + +/** + * 事件流的读取:发给我的事件 + 我所在公会的广播。`vk_events.id` 是游标。 + * @param {{ db: import('knex').Knex }} deps + */ +export function createEvents({ db }) { + if (typeof db !== 'function') throw new Error('createEvents: db is required') + + /** + * @param {import('./agents.js').Agent} agent + * @param {number} since + * @returns {Promise} + */ + async function after(agent, since) { + return db(EVENTS) + .where('id', '>', since) + .where((q) => q.where({ target_agent_id: agent.id }).orWhere((g) => g.whereNull('target_agent_id').where({ guild_id: agent.guild_id }))) + .orderBy('id', 'asc') + .limit(PAGE) + .select('id', 'type', 'task_id', 'created_at') + } + + /** + * `/home` 用:返回上次已读游标之后的事件,并把游标推进到本次返回的最后一条。 + * @param {object} _ctx + * @param {{ agent: import('./agents.js').Agent }} input + * @returns {Promise} + */ + async function pullUnread(_ctx, { agent }) { + const row = await db('vk_agents').where({ id: agent.id }).first('events_cursor') + const events = await after(agent, row.events_cursor) + if (events.length > 0) await db('vk_agents').where({ id: agent.id }).where('events_cursor', '<', events[events.length - 1].id).update({ events_cursor: events[events.length - 1].id }) + return events + } + + /** + * `/events?since=` 用:补取,不推进游标。 + * @param {object} _ctx + * @param {{ agent: import('./agents.js').Agent, since: number }} input + * @returns {Promise} + */ + async function listSince(_ctx, { agent, since }) { + return after(agent, since) + } + + return { pullUnread, listSince } +} diff --git a/packages/voko-exchange/src/index.js b/packages/voko-exchange/src/index.js index 752b633..640c37d 100644 --- a/packages/voko-exchange/src/index.js +++ b/packages/voko-exchange/src/index.js @@ -1,13 +1,17 @@ /** * `@voko/exchange` —— VOKO 的任务交换:公会、Agent、任务、检查点、提交、验收、评论、求助、事件与定时任务。 * - * 当前有表结构、Agent 与公会的读写、任务构造点与状态机纯函数;任务 service 在 W2–W3 加入。 + * 当前有表结构、Agent 与公会的读写、任务状态机与发布 / 认领 / 报名 / 修改 / 撤回 / 评论、事件游标、定时任务执行;检查点、提交与验收在 W3 加入。 * * 边界:不读 `process.env`、不建连接;db 由 composition root 注入。 */ export { createAgents, agentFromRow, AGENT_STATUSES, AGENT_ROLES } from './agents.js' export { createGuilds } from './guilds.js' +export { createTasks } from './tasks.js' +export { createEvents } from './events.js' +export { createJobs } from './jobs.js' +export { rarityForReward, maxRarityForLevel } from './rarity.js' export { taskFromRow, TASK_STATUSES, TASK_TEMPLATES } from './task.js' export { transition, publishStatus, TASK_ACTIONS, MAX_ACTIVE_HOLDS } from './transition.js' export { default as exchangeDbSchema } from './db/tables/index.cjs' diff --git a/packages/voko-exchange/src/jobs.js b/packages/voko-exchange/src/jobs.js new file mode 100644 index 0000000..08fef61 --- /dev/null +++ b/packages/voko-exchange/src/jobs.js @@ -0,0 +1,82 @@ +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, + * }, + * 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} + */ + 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 (TASK_ACTION_KINDS.includes(job.kind) || WARN_KINDS.includes(job.kind)) { + if (job.ref_id === null) throw new Error(`vk_jobs.ref_id missing for ${job.kind}, id: ${job.id}`) + if (WARN_KINDS.includes(job.kind)) await tasks.warn(ctx, { task_id: job.ref_id, kind: job.kind }) + else await tasks.actAsWorker(ctx, { task_id: job.ref_id, action: { type: job.kind } }) + return + } + // 其余类型(自动确认、仲裁、Webhook、日报、对账)在 W3–W4 接入。 + throw new Error(`Unhandled job kind: ${job.kind}, id: ${job.id}`) + } + + return { runDue } +} diff --git a/packages/voko-exchange/src/rarity.js b/packages/voko-exchange/src/rarity.js new file mode 100644 index 0000000..9ac03aa --- /dev/null +++ b/packages/voko-exchange/src/rarity.js @@ -0,0 +1,23 @@ +// ponytail: 赏金区间与等级上限先写成常量;PRD 说「由运营配置」,要在后台改时换成配置表(W5 等级系统一起做)。 +/** 稀有度 n(1–10)的赏金下限;低于第二档的都是 1。 */ +const RARITY_MIN_REWARD = Object.freeze([1, 50, 100, 200, 400, 800, 1600, 3200, 6400, 12800]) + +/** + * 稀有度按赏金区间自动得出:1 最小,10 最大。 + * @param {number} reward 正整数 + * @returns {number} + */ +export function rarityForReward(reward) { + if (!Number.isSafeInteger(reward) || reward < 1) throw new Error(`rarityForReward: reward invalid: ${reward}`) + return RARITY_MIN_REWARD.findLastIndex((min) => reward >= min) + 1 +} + +/** + * 等级决定可接的稀有度上限:1 级可接到 3,每升一级 +1,封顶 10。 + * @param {number} level 从 1 起 + * @returns {number} + */ +export function maxRarityForLevel(level) { + if (!Number.isSafeInteger(level) || level < 1) throw new Error(`maxRarityForLevel: level invalid: ${level}`) + return Math.min(10, level + 2) +} diff --git a/packages/voko-exchange/src/tasks.js b/packages/voko-exchange/src/tasks.js new file mode 100644 index 0000000..6440775 --- /dev/null +++ b/packages/voko-exchange/src/tasks.js @@ -0,0 +1,391 @@ +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, + * escrow: { + * hold(ctx: object, trx: import('knex').Knex, input: { task_id: string, payer_wallet: string, amount: number }): Promise, + * topupExists(ctx: object, trx: import('knex').Knex, input: { task_id: string, idempotency_key: string }): Promise, + * topup(ctx: object, trx: import('knex').Knex, input: { task_id: string, amount: number, idempotency_key: string }): Promise, + * refund(ctx: object, trx: import('knex').Knex, input: { task_id: string }): Promise, + * }, + * }} 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} + */ +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} + */ +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 } +} diff --git a/packages/voko-exchange/src/transition.js b/packages/voko-exchange/src/transition.js index 635a5da..6412ff0 100644 --- a/packages/voko-exchange/src/transition.js +++ b/packages/voko-exchange/src/transition.js @@ -8,6 +8,7 @@ * 重复报名(vk_task_entries UNIQUE)、转让请求是否 pending 且指向 actor、验收 checks 与清单的覆盖关系。 */ +import { rarityForReward } from './rarity.js' import { TASK_STATUSES } from './task.js' /** 除发布(新建,见 `publishStatus`)外的全部动作。 */ @@ -160,7 +161,8 @@ const TRANSITIONS = { if (!Number.isSafeInteger(action.add_reward) || action.add_reward < 0) throw new Error(`edit.add_reward invalid: ${action.add_reward}, id: ${task.id}`) if (action.add_reward === 0 && action.deadline_at === null) return fail('rule_violation', 'Nothing to change: add reward or extend the deadline.') if (action.deadline_at !== null && action.deadline_at <= task.deadline_at) return fail('rule_violation', 'The deadline can only be extended.') - const next = { ...task, reward: task.reward + action.add_reward, deadline_at: action.deadline_at === null ? task.deadline_at : action.deadline_at } + const reward = task.reward + action.add_reward + const next = { ...task, reward, rarity: rarityForReward(reward), deadline_at: action.deadline_at === null ? task.deadline_at : action.deadline_at } return ok(next, [ ...(action.add_reward === 0 ? [] : [/** @type {Effect} */ ({ type: 'escrow_topup', amount: action.add_reward })]), ...(action.deadline_at === null ? [] : openJobs(next)), @@ -217,6 +219,7 @@ const TRANSITIONS = { enter(state, _action, actor) { const refused = takerRefusal(state, actor) if (refused !== null) return refused + if (/** @type {AgentActor} */ (actor).is_entrant) return fail('state_conflict', 'You have already entered this contest.') return ok(state.task, [{ type: 'entry_add', agent_id: /** @type {AgentActor} */ (actor).agent_id }]) }, submit({ task }, _action, actor) { diff --git a/packages/voko-exchange/test/rarity.test.js b/packages/voko-exchange/test/rarity.test.js new file mode 100644 index 0000000..3d7aa95 --- /dev/null +++ b/packages/voko-exchange/test/rarity.test.js @@ -0,0 +1,17 @@ +import { describe, expect, it } from 'vitest' +import { maxRarityForLevel, rarityForReward } from '../src/rarity.js' + +describe('稀有度区间与等级上限的边界值', () => { + it.each([[1, 1], [49, 1], [50, 2], [99, 2], [100, 3], [12799, 9], [12800, 10], [1_000_000, 10]])('reward %i → rarity %i', (reward, rarity) => { + expect(rarityForReward(reward)).toBe(rarity) + }) + + it.each([[1, 3], [2, 4], [8, 10], [50, 10]])('level %i → 上限 %i', (level, max) => { + expect(maxRarityForLevel(level)).toBe(max) + }) + + it('非法输入直接 throw', () => { + expect(() => rarityForReward(0)).toThrow(/reward invalid/) + expect(() => maxRarityForLevel(0)).toThrow(/level invalid/) + }) +}) diff --git a/packages/voko-exchange/test/transition.test.js b/packages/voko-exchange/test/transition.test.js index 174ede5..a6ab07c 100644 --- a/packages/voko-exchange/test/transition.test.js +++ b/packages/voko-exchange/test/transition.test.js @@ -207,12 +207,16 @@ describe('SM-01 副作用与平台规则', () => { it('修改只能加赏金或延长截止', () => { const edited = run(stateFor('open'), 'edit') - expect(edited.task).toMatchObject({ reward: 150, deadline_at: FUTURE }) + expect(edited.task).toMatchObject({ reward: 150, rarity: 3, deadline_at: FUTURE }) expect(effectTypes(edited)).toEqual(['escrow_topup', 'job:open_expire']) expect(run(stateFor('open'), 'edit', publisher, { add_reward: 0, deadline_at: null })).toMatchObject({ ok: false, code: 'rule_violation' }) expect(run(stateFor('open', 'piecework', { deadline_at: FUTURE }), 'edit', publisher, { add_reward: 0, deadline_at: PAST })).toMatchObject({ ok: false, code: 'rule_violation' }) }) + it('重复报名 → state_conflict', () => { + expect(run(stateFor('in_contest'), 'enter', { ...entrant, agent_id: 'ag_entrant' })).toMatchObject({ ok: false, code: 'state_conflict' }) + }) + it('争议只能由发布方或接单方发起', () => { expect(run(stateFor('submitted'), 'dispute', holder).task.status).toBe('disputed') expect(run(stateFor('reviewing'), 'dispute', entrant).task.status).toBe('disputed') diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3e88489..7981a0e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -63,6 +63,9 @@ importers: '@drama/observability': specifier: workspace:* version: link:../../packages/observability + '@drama/ratelimit': + specifier: workspace:* + version: link:../../packages/ratelimit '@drama/server-kit': specifier: workspace:* version: link:../../packages/server-kit @@ -292,7 +295,11 @@ importers: specifier: ^2.1.4 version: 2.1.9(@types/node@24.13.2)(jsdom@25.0.1)(sass@1.101.0) - packages/voko-clearing: {} + packages/voko-clearing: + dependencies: + '@drama/billing': + specifier: workspace:* + version: link:../billing packages/voko-exchange: devDependencies: diff --git a/tools/scripts/verify-local.sh b/tools/scripts/verify-local.sh index 517c60c..8818114 100755 --- a/tools/scripts/verify-local.sh +++ b/tools/scripts/verify-local.sh @@ -1,5 +1,5 @@ #!/usr/bin/env bash -# 本地一键验收:起 Docker MySQL → 空库则建表 → 建本地管理员 → 启动 API → 跑骨架与 F0 真实链路脚本 → 停 API。 +# 本地一键验收:起 Docker MySQL → 空库则建表 → 建本地管理员 → 启动 API 与 worker → 跑全部真实链路脚本 → 停进程。 # 用法:VOKO_TEST_EMAIL=you@gmail.com bash tools/scripts/verify-local.sh set -euo pipefail cd "$(dirname "$0")/../.." @@ -18,7 +18,9 @@ pnpm dev:user admin@example.test admin123 node apps/voko-server/src/main.js > tmp-voko-server.log 2>&1 & SERVER_PID=$! -trap 'kill $SERVER_PID 2>/dev/null || true' EXIT +node apps/voko-server/src/voko-worker-main.js > tmp-voko-worker.log 2>&1 & +WORKER_PID=$! +trap 'kill $SERVER_PID $WORKER_PID 2>/dev/null || true' EXIT for _ in $(seq 1 30); do curl -sf http://localhost:9201/healthz >/dev/null && break sleep 1 @@ -27,5 +29,7 @@ curl -sf http://localhost:9201/healthz >/dev/null || { echo "API 未启动,日 VOKO_ADMIN_EMAIL=admin@example.test VOKO_ADMIN_PASSWORD=admin123 \ pnpm --filter @voko/server verify:skeleton -VOKO_ADMIN_EMAIL=admin@example.test VOKO_ADMIN_PASSWORD=admin123 \ - pnpm --filter @voko/server verify:f0 +export VOKO_ADMIN_EMAIL=admin@example.test VOKO_ADMIN_PASSWORD=admin123 +for scenario in f0 f1 f2 concurrency timers; do + pnpm --filter @voko/server "verify:$scenario" +done