_lib.mjs
13.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
/**
* 场景脚本公共库：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 { rm } from 'node:fs/promises'
import { fileURLToPath } from 'node:url'
import { createBillingRepo, createLedger } from '@drama/billing'
import { createDb } from '@drama/db'
import { verifyConservation } from '@voko/clearing'
import { databaseUrlFromArgs } from '../../../src/db/env-file.js'
const api = (process.env.VOKO_API || 'http://localhost:9201').replace(/\/$/, '')
// 与服务端 env.js 的缺省值一致：本地交付文件的存储目录。
const filesDir = process.env.VOKO_FILES_DIR || fileURLToPath(new URL('../../../data/files', import.meta.url))
/**
* @param {string} name 场景名，用于测试邮箱与公会 slug
* @param {(world: ReturnType<typeof createWorld>) => Promise<void>} 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', handle: `${name}-${tag}-${run}`.slice(0, 32) } })
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
}
/** `settings` 可覆盖 require_lead_approval 与四个时间规则（分钟级 / 秒级的公会用于时间类用例）。 */
async function createGuild(tag, leadUserId, settings = {}) {
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, ...settings } })
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, role = 'member') {
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, role } })
check(`Agent ${tag}：认领`, claimed.status === 200 && claimed.json.agent.status === 'active', claimed)
return { id: data.agent_id, key: data.api_key }
}
/** 只注册不认领的 Agent（用于「未认领 / 无权认领」的用例）；结束时一并清理。 */
async function registerAgent(tag) {
const registered = await call('POST', '/v1/agents/register', { body: { name: `${name}-${tag}-${run}`, skills: [] } })
check(`Agent ${tag}：注册`, registered.status === 200, registered)
agentIds.push(registered.json.data.agent_id)
return registered.json.data
}
/** 运营给 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) {
const fileIds = (await db('vk_files').whereIn('task_id', taskIds).select('id')).map((row) => row.id)
for (const id of fileIds) await rm(`${filesDir}/${id}`, { force: true })
await db('vk_files').whereIn('task_id', taskIds).del()
await db('vk_review_checks').whereIn('review_id', db('vk_reviews').whereIn('task_id', taskIds).select('id')).del()
for (const table of ['vk_task_acceptance_items', 'vk_task_holds', 'vk_task_entries', 'vk_comments', 'vk_checkpoints', 'vk_submissions', 'vk_reviews', 'vk_disputes', 'vk_transfers', 'vk_help_requests', 'vk_tips', '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_jobs').where({ kind: 'help_notify' }).whereNotIn('ref_id', db('vk_help_requests').select('id')).del()
await db('vk_tasks').whereIn('id', taskIds).del()
}
if (guildIds.length > 0) {
await db('vk_webhook_deliveries').whereIn('event_id', db('vk_events').whereIn('guild_id', guildIds).select('id')).del()
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_titles').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_jobs').where({ kind: 'daily_digest' }).whereIn('ref_id', userIds).del()
await db('vk_access_requests').whereIn('user_id', userIds).del()
await db('vk_owner_settings').whereIn('user_id', userIds).del()
await db('vk_owner_handles').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()
}
}
/** 轮询任务详情直到满足条件（等真实 worker 执行），最多 `seconds` 秒；返回最后一次读到的任务。 */
async function waitForTask(agent, taskId, predicate, seconds = 40) {
let task = null
for (let i = 0; i < seconds; i += 1) {
task = (await call('GET', `/v1/tasks/${taskId}`, { token: agent.key })).json.data
if (predicate(task)) return task
await new Promise((resolve) => setTimeout(resolve, 1000))
}
return task
}
const wallet = async (agent) => (await call('GET', '/v1/wallet', { token: agent.key })).json.data
const SUBMISSION = { summary: 'Translated; see the doc.', artifacts: [{ type: 'link', url: 'https://example.com/doc/1', note: 'translation' }] }
const submit = (agent, taskId, summary = SUBMISSION.summary) => call('POST', `/v1/tasks/${taskId}/submit`, { token: agent.key, body: { ...SUBMISSION, summary } })
/** 上传一个交付文件（原始字节），返回接口响应。 */
async function uploadFile(agent, taskId, filename, content) {
const response = await fetch(`${api}/v1/tasks/${taskId}/files?filename=${encodeURIComponent(filename)}`, { method: 'POST', headers: { authorization: `Bearer ${agent.key}`, 'content-type': 'application/octet-stream' }, body: content })
return { status: response.status, json: await response.json() }
}
/** 下载一个交付文件；成功时返回正文文本。 */
async function downloadFile(agent, fileId) {
const response = await fetch(`${api}/v1/files/${fileId}`, { headers: { authorization: `Bearer ${agent.key}` } })
return { status: response.status, disposition: response.headers.get('content-disposition'), text: response.status === 200 ? await response.text() : null }
}
const claim = (agent, taskId) => call('POST', `/v1/tasks/${taskId}/claim`, { token: agent.key })
/** 按清单逐条验收；`failing` 是不通过的清单项 id。 */
const review = (agent, taskId, action, ids, failing = [], extra = {}) => call('POST', `/v1/tasks/${taskId}/review`, {
token: agent.key,
body: { action, checks: ids.map((id) => ({ id, pass: !failing.includes(id), reason: failing.includes(id) ? 'Not met yet.' : 'Looks good.' })), ...extra },
})
return { db, results, check, call, admin, waitForEmail, createOwner, createGuild, createAgent, registerAgent, grantBudget, publish, waitForTask, wallet, submit, claim, review, uploadFile, downloadFile, cleanup }
}